From 457365d3d18e5c1ec86a61c63c9df9e36e41f573 Mon Sep 17 00:00:00 2001 From: andrejsimic Date: Thu, 15 Oct 2026 18:56:57 +0200 Subject: [PATCH] Poll on task availability notices (4.6.0) A transport can now announce when tasks of a type may be waiting (ITaskAvailabilityNotifications, in ConductorSharp.Client). UseTaskAvailabilityNotifications(fallbackPollInterval) switches to an execution manager that polls a type when it is announced, keeps polling it until a poll comes back short, and otherwise polls only on the fallback interval. An idle worker sends nothing else. Maestro's NATS transport implements the interface. - The per-task handling both execution managers carried a copy of moved to PolledTaskRunner, which all three use. - The background service treats an OperationCanceledException on stop as a stop. The new manager waits on a semaphore and a channel, which throw that rather than TaskCanceledException. --- README.md | 15 + SKILL.md | 10 + .../ConductorSharp.Client.csproj | 2 +- .../Service/ITaskAvailabilityNotifications.cs | 28 ++ .../ConductorSharp.Engine.csproj | 2 +- src/ConductorSharp.Engine/ExecutionManager.cs | 158 +------ .../Extensions/ConductorSharpBuilder.cs | 18 + .../Extensions/IExecutionManagerBuilder.cs | 15 + .../NotifiedExecutionManager.cs | 320 +++++++++++++ .../NotifiedPollingConfig.cs | 14 + src/ConductorSharp.Engine/PolledTaskRunner.cs | 187 ++++++++ .../WorkflowEngineBackgroundService.cs | 4 +- .../TypePollSpreadingExecutionManager.cs | 158 +------ ...ctorSharp.KafkaCancellationNotifier.csproj | 2 +- .../ConductorSharp.Patterns.csproj | 2 +- .../ConductorSharp.Toolkit.csproj | 2 +- .../Unit/NotifiedExecutionManagerTests.cs | 441 ++++++++++++++++++ 17 files changed, 1066 insertions(+), 312 deletions(-) create mode 100644 src/ConductorSharp.Client/Service/ITaskAvailabilityNotifications.cs create mode 100644 src/ConductorSharp.Engine/NotifiedExecutionManager.cs create mode 100644 src/ConductorSharp.Engine/NotifiedPollingConfig.cs create mode 100644 src/ConductorSharp.Engine/PolledTaskRunner.cs create mode 100644 test/ConductorSharp.Engine.Tests/Unit/NotifiedExecutionManagerTests.cs diff --git a/README.md b/README.md index 2764a1f8..35bfe110 100644 --- a/README.md +++ b/README.md @@ -706,6 +706,21 @@ public class MyController( .UseConstantPollTimingStrategy() ``` +### Polling on task availability notices (Maestro) + +A transport that can announce when tasks of a type may be waiting registers +`ITaskAvailabilityNotifications`. Maestro's NATS transport does. With it, the worker polls a type +when its notice arrives and keeps polling until the queue is empty, instead of reading the queue +sizes on every sleep interval. An idle worker sends nothing but a fallback poll, which picks up +tasks whose notice was lost: + +```csharp +.AddExecutionManager(...) +.UseTaskAvailabilityNotifications(fallbackPollInterval: 30000) +``` + +Conductor's REST API has no such notices, so this only works with a transport that provides them. + ### Worker Task Registration Register standalone tasks without workflow: diff --git a/SKILL.md b/SKILL.md index b65779d4..3af60dda 100644 --- a/SKILL.md +++ b/SKILL.md @@ -749,6 +749,16 @@ public class MyController( .UseBetaExecutionManager() // Uses TypePollSpreadingExecutionManager ``` +### Polling on Task Availability Notices (Maestro) + +```csharp +.AddExecutionManager(...) +.UseTaskAvailabilityNotifications(fallbackPollInterval: 30000) // Polls a type when the transport announces it +``` + +Needs a transport that registers `ITaskAvailabilityNotifications`, such as Maestro's NATS transport +(`AddMaestroNatsWorkers`). An idle worker then sends only the fallback poll. + ### Patterns Package ```csharp diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 946018b3..c5e1072f 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 4.5.0 + 4.6.0 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.Client/Service/ITaskAvailabilityNotifications.cs b/src/ConductorSharp.Client/Service/ITaskAvailabilityNotifications.cs new file mode 100644 index 00000000..b8013def --- /dev/null +++ b/src/ConductorSharp.Client/Service/ITaskAvailabilityNotifications.cs @@ -0,0 +1,28 @@ +namespace ConductorSharp.Client.Service +{ + /// + /// Tells workers when tasks of a type may be waiting, so they poll on notice instead of on a timer. + /// + /// + /// A transport implements this when the server can push such notices. Maestro does over NATS. + /// Conductor's REST API has nothing like it, so the HTTP clients do not implement it. + /// + public interface ITaskAvailabilityNotifications + { + /// + /// Starts listening for notices about , and calls + /// with the task type of each one. Disposing the result stops + /// listening. + /// + /// + /// Returns once the subscription is in place, so a notice sent after that reaches the callback. + /// The callback runs on the transport's own thread and has to return at once. A transport that + /// may have missed notices, as one that just reconnected, calls it for every type. + /// + Task SubscribeAsync( + IReadOnlyCollection taskTypes, + Action onAvailable, + CancellationToken cancellationToken = default + ); + } +} diff --git a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index d2e48908..717b042d 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 4.5.0 + 4.6.0 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 dcf21cf6..a29cf87a 100644 --- a/src/ConductorSharp.Engine/ExecutionManager.cs +++ b/src/ConductorSharp.Engine/ExecutionManager.cs @@ -28,25 +28,21 @@ internal class ExecutionManager : IExecutionManager private readonly WorkerSetConfig _configuration; private readonly ILogger _logger; private readonly ITaskService _taskManager; - private readonly IExternalPayloadService _externalPayloadService; private readonly IEnumerable _registeredWorkers; - private readonly IServiceScopeFactory _lifetimeScopeFactory; private readonly IPollTimingStrategy _pollTimingStrategy; private readonly IPollOrderStrategy _pollOrderStrategy; - private readonly ICancellationNotifier _cancellationNotifier; private readonly TaskQueuePollingService _taskQueuePollingService; + private readonly PolledTaskRunner _runner; public ExecutionManager( WorkerSetConfig options, ILogger logger, ITaskService taskService, IEnumerable workerMappings, - IExternalPayloadService externalPayloadService, - IServiceScopeFactory lifetimeScope, IPollTimingStrategy pollTimingStrategy, IPollOrderStrategy pollOrderStrategy, - ICancellationNotifier cancellationNotifier, - TaskQueuePollingService taskQueuePollingService + TaskQueuePollingService taskQueuePollingService, + PolledTaskRunner runner ) { _configuration = options; @@ -54,12 +50,10 @@ TaskQueuePollingService taskQueuePollingService _logger = logger; _taskManager = taskService; _registeredWorkers = workerMappings; - _lifetimeScopeFactory = lifetimeScope; _pollTimingStrategy = pollTimingStrategy; _pollOrderStrategy = pollOrderStrategy; - _cancellationNotifier = cancellationNotifier; - _externalPayloadService = externalPayloadService; _taskQueuePollingService = taskQueuePollingService; + _runner = runner; } public async Task StartAsync(CancellationToken cancellationToken) @@ -103,20 +97,6 @@ private string GetQueueTaskName(TaskToWorker taskToWorker) return taskToWorker.TaskName; } - private static Type GetInputType(Type workerType) - { - var interfaces = workerType - .GetInterfaces() - .Where(a => a.IsGenericType && a.GetGenericTypeDefinition() == typeof(ITaskRequestHandler<,>)) - .First(); - var genericArguments = interfaces.GetGenericArguments(); - - var inputType = genericArguments[0]; - var outputType = genericArguments[1]; - - return inputType; - } - private async Task PollAndHandle(TaskToWorker scheduledWorker, CancellationToken cancellationToken) { Client.Generated.Task pollResponse; @@ -151,135 +131,7 @@ private async Task PollAndHandle(TaskToWorker scheduledWorker, CancellationToken if (pollResponse == null) return; - await ProcessPolledTask(pollResponse, workerId, scheduledWorker, cancellationToken); - } - - private async Task ProcessPolledTask( - Client.Generated.Task pollResponse, - string workerId, - TaskToWorker scheduledWorker, - CancellationToken cancellationToken - ) - { - using var tokenHolder = _cancellationNotifier.GetCancellationToken(pollResponse.TaskId, cancellationToken); - - try - { - if (!string.IsNullOrEmpty(pollResponse.ExternalInputPayloadStoragePath)) - { - _logger.LogDebug("Fetching storage {location}", pollResponse.ExternalInputPayloadStoragePath); - // TODO: Check what the operation and payload type are - var externalStorageLocation = await _taskManager.GetExternalStorageLocationAsync( - pollResponse.ExternalInputPayloadStoragePath, - "READ", - "TASK_INPUT", - cancellationToken - ); - - // TODO: iffy - var file = await _externalPayloadService.GetExternalStorageDataAsync(externalStorageLocation.Path, tokenHolder.CancellationToken); - - using TextReader textReader = new StreamReader(file.Stream); - var json = await textReader.ReadToEndAsync(); - - pollResponse.InputData = JsonConvert.DeserializeObject>( - json, - ConductorConstants.IoJsonSerializerSettings - ); - } - - var inputType = GetInputType(scheduledWorker.TaskType); - var inputData = SerializationHelper.DictonaryToObject(inputType, pollResponse.InputData, ConductorConstants.IoJsonSerializerSettings); - // Poll response data can be huge (if read from external storage) - // We can save memory by not holding reference to pollResponse.InputData after it is parsed - pollResponse.InputData = null; - - using var scope = _lifetimeScopeFactory.CreateScope(); - - var context = scope.ServiceProvider.GetService(); - var mediator = scope.ServiceProvider.GetRequiredService(); - - if (context != null) - { - context.WorkflowName = pollResponse.WorkflowType; - context.TaskName = pollResponse.TaskDefName; - context.TaskReferenceName = pollResponse.ReferenceTaskName; - context.WorkflowId = pollResponse.WorkflowInstanceId; - context.CorrelationId = pollResponse.CorrelationId; - context.Context = ConductorSharpExecutionContext.CopyContext(pollResponse.Context); - context.TaskId = pollResponse.TaskId; - context.WorkerId = workerId; - } - - var response = await mediator.Send(inputData, tokenHolder.CancellationToken); - - await _taskManager.UpdateAsync( - new TaskResult - { - TaskId = pollResponse.TaskId, - Status = TaskResultStatus.COMPLETED, - OutputData = SerializationHelper.ObjectToDictionary(response, ConductorConstants.IoJsonSerializerSettings), - WorkflowInstanceId = pollResponse.WorkflowInstanceId - }, - tokenHolder.CancellationToken - ); - } - catch (OperationCanceledException) when (tokenHolder.IsCancellationRequestedByNotifier) - { - _logger.LogWarning( - "Polled task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) is cancelled", - pollResponse.TaskDefName, - pollResponse.TaskId, - pollResponse.WorkflowType, - pollResponse.WorkflowInstanceId - ); - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) // This is fine since we know cancellationToken comes from background service - { - _logger.LogWarning( - "Cancelling task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) due to background service shutdown", - pollResponse.TaskDefName, - pollResponse.TaskId, - pollResponse.WorkflowType, - pollResponse.WorkflowInstanceId - ); - } - catch (Exception exception) - { - _logger.LogError( - "{@Exception} while executing {Task} as part of {Workflow} with id {WorkflowId}", - exception, - pollResponse.TaskDefName, - pollResponse.WorkflowType, - pollResponse.WorkflowInstanceId - ); - - // A worker may throw a StructuredErrorException to attach a sanitized, stable classification to the - // failed task's output. Plain exceptions keep producing only error_message, preserving backward compatibility. - var errorMessage = new ErrorOutput { ErrorMessage = exception.Message, StructuredError = StructuredError.FromException(exception) }; - - // TODO: We should verify that this is alright, it is possible that when executed concurrently, - // the updates caused by LogAsync will be discarded because the call of UpdateAsync(TaskResult...) - // sets the logs to null. Not sure how this is implemented in the backend, also, would have expected this to be a - // PUT or PATCH request, but by specs it is POST - await Task.WhenAll( - [ - _taskManager.UpdateAsync( - new TaskResult - { - TaskId = pollResponse.TaskId, - Status = TaskResultStatus.FAILED, - ReasonForIncompletion = exception.Message, - OutputData = SerializationHelper.ObjectToDictionary(errorMessage, ConductorConstants.IoJsonSerializerSettings), - WorkflowInstanceId = pollResponse?.WorkflowInstanceId - }, - tokenHolder.CancellationToken - ), - _taskManager.LogAsync(pollResponse.TaskId, exception.Message, tokenHolder.CancellationToken), - _taskManager.LogAsync(pollResponse.TaskId, exception.StackTrace, tokenHolder.CancellationToken) - ] - ); - } + await _runner.RunAsync(pollResponse, workerId, scheduledWorker, cancellationToken); } } } diff --git a/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs b/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs index 6c1b670d..dedf1f00 100644 --- a/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs +++ b/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs @@ -47,6 +47,8 @@ params Assembly[] handlerAssemblies Builder.AddSingleton(); + Builder.AddSingleton(); + Builder.AddScoped(); Builder.AddTransient(); @@ -70,6 +72,22 @@ public IExecutionManagerBuilder UseBetaExecutionManager() return this; } + public IExecutionManagerBuilder UseTaskAvailabilityNotifications(int fallbackPollInterval = 30000) + { + if (fallbackPollInterval <= 0) + { + throw new ArgumentOutOfRangeException( + nameof(fallbackPollInterval), + fallbackPollInterval, + "The fallback poll interval must be positive." + ); + } + + Builder.AddSingleton(new NotifiedPollingConfig { FallbackPollInterval = fallbackPollInterval }); + Builder.AddSingleton(); + return this; + } + public IExecutionManagerBuilder AddPipelines(Action behaviorBuilder) { var pipelineBuilder = new PipelineBuilder(Builder); diff --git a/src/ConductorSharp.Engine/Extensions/IExecutionManagerBuilder.cs b/src/ConductorSharp.Engine/Extensions/IExecutionManagerBuilder.cs index 762f5c32..d834e532 100644 --- a/src/ConductorSharp.Engine/Extensions/IExecutionManagerBuilder.cs +++ b/src/ConductorSharp.Engine/Extensions/IExecutionManagerBuilder.cs @@ -13,5 +13,20 @@ IExecutionManagerBuilder SetHealthCheckService() where T : IConductorSharpHealthService; IExecutionManagerBuilder UseConstantPollTimingStrategy(); IExecutionManagerBuilder UseBetaExecutionManager(); + + /// + /// Polls a task type when the transport announces that tasks of it may be waiting, and keeps + /// polling it until its queue is empty, instead of reading the queue sizes on every sleep + /// interval. An idle worker then sends nothing but the fallback poll. + /// + /// + /// Needs a transport that registers , + /// such as Maestro's NATS transport. Conductor's REST API has no such notices. + /// + /// + /// How often every task type is polled anyway, in milliseconds. It picks up tasks whose + /// notice was lost, so it bounds how long such a task waits. + /// + IExecutionManagerBuilder UseTaskAvailabilityNotifications(int fallbackPollInterval = 30000); } } diff --git a/src/ConductorSharp.Engine/NotifiedExecutionManager.cs b/src/ConductorSharp.Engine/NotifiedExecutionManager.cs new file mode 100644 index 00000000..eb30a4dd --- /dev/null +++ b/src/ConductorSharp.Engine/NotifiedExecutionManager.cs @@ -0,0 +1,320 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; +using ConductorSharp.Client; +using ConductorSharp.Client.Generated; +using ConductorSharp.Client.Service; +using ConductorSharp.Engine.Model; +using Microsoft.Extensions.Logging; +using Task = System.Threading.Tasks.Task; + +namespace ConductorSharp.Engine +{ + /// + /// Polls a task type when the transport announces that tasks of it may be waiting, and keeps + /// polling it until its queue is empty. Every type is also polled on a slow fallback interval. + /// + /// + /// The other execution managers read the queue sizes on every sleep interval, whether anything + /// was scheduled or not. This one sends nothing while it is idle, apart from the fallback poll, + /// which picks up tasks whose notice the transport lost. + /// + /// A notice marks its type pending. The loop polls each pending type in batches as large as the + /// free worker slots, and waits for a slot when every worker is busy. It clears the type once a + /// poll returns fewer tasks than it asked for, which means the queue was empty. A notice that + /// arrives while a poll is out keeps the type pending. That poll may have read the queue before + /// the commit the notice stands for, and clearing the type would leave that task for the + /// fallback poll. + /// + internal class NotifiedExecutionManager : IExecutionManager + { + /// + /// The most tasks one batch poll asks for. Conductor and Maestro refuse more than 100. + /// + private const int MaxBatchSize = 100; + + private static readonly TimeSpan MaxSubscribeRetryDelay = TimeSpan.FromSeconds(30); + + private readonly WorkerSetConfig _configuration; + private readonly TimeSpan _fallbackPollInterval; + private readonly ILogger _logger; + private readonly ITaskService _taskService; + private readonly ITaskAvailabilityNotifications _notifications; + private readonly IReadOnlyList _registeredWorkers; + private readonly PolledTaskRunner _runner; + private readonly SemaphoreSlim _slots; + + /// + /// Wakes the loop when a notice arrives. One wake waiting is enough, because the loop looks + /// at every pending type when it wakes. + /// + private readonly Channel _wake = Channel.CreateBounded( + new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropWrite } + ); + + public NotifiedExecutionManager( + WorkerSetConfig configuration, + NotifiedPollingConfig pollingConfig, + ILogger logger, + ITaskService taskService, + ITaskAvailabilityNotifications notifications, + IEnumerable workerMappings, + PolledTaskRunner runner + ) + { + _configuration = configuration; + _fallbackPollInterval = TimeSpan.FromMilliseconds(pollingConfig.FallbackPollInterval); + _logger = logger; + _taskService = taskService; + _notifications = notifications; + _registeredWorkers = workerMappings.ToList(); + _runner = runner; + _slots = new SemaphoreSlim(configuration.MaxConcurrentWorkers); + } + + public async Task StartAsync(CancellationToken cancellationToken) + { + var queues = _registeredWorkers.Select(worker => new WorkerQueue(worker)).ToList(); + var queuesByTaskType = queues.GroupBy(queue => queue.Worker.TaskName).ToDictionary(group => group.Key, group => group.ToList()); + + await using var subscription = await SubscribeAsync(queuesByTaskType, cancellationToken); + + // Tasks queued before the subscription was in place had their notices already. + MarkPending(queues); + var nextFallbackPoll = DateTimeOffset.UtcNow + _fallbackPollInterval; + + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + + if (DateTimeOffset.UtcNow >= nextFallbackPoll) + { + MarkPending(queues); + nextFallbackPoll = DateTimeOffset.UtcNow + _fallbackPollInterval; + } + + var pending = queues.Where(queue => queue.IsPending).ToList(); + if (pending.Count == 0) + { + await WaitForNoticeAsync(nextFallbackPoll - DateTimeOffset.UtcNow, cancellationToken); + continue; + } + + // Shuffled, so a type that always has work cannot keep the slots from the others. + foreach (var queue in pending.OrderBy(_ => Random.Shared.Next())) + { + await PollAsync(queue, cancellationToken); + } + } + } + + /// + /// Subscribes to the notices of every registered type, retrying until the transport accepts. + /// Nothing is polled before that, since a poll's answer is only useful while notices arrive. + /// + private async Task SubscribeAsync( + Dictionary> queuesByTaskType, + CancellationToken cancellationToken + ) + { + var delay = TimeSpan.FromSeconds(1); + + while (true) + { + try + { + return await _notifications.SubscribeAsync( + queuesByTaskType.Keys.ToList(), + taskType => OnNotice(queuesByTaskType, taskType), + cancellationToken + ); + } + catch (Exception exception) when (!cancellationToken.IsCancellationRequested) + { + _logger.LogWarning(exception, "Could not subscribe to task availability notices; retrying in {Delay}", delay); + + await Task.Delay(delay, cancellationToken); + delay = TimeSpan.FromTicks(Math.Min(delay.Ticks * 2, MaxSubscribeRetryDelay.Ticks)); + } + } + } + + private void OnNotice(Dictionary> queuesByTaskType, string taskType) + { + if (!queuesByTaskType.TryGetValue(taskType, out var queues)) + return; + + foreach (var queue in queues) + queue.MarkPending(); + + _wake.Writer.TryWrite(true); + } + + private static void MarkPending(IEnumerable queues) + { + foreach (var queue in queues) + queue.MarkPending(); + } + + /// + /// Waits for a notice, or until passes and the fallback poll is due. + /// + private async Task WaitForNoticeAsync(TimeSpan timeout, CancellationToken cancellationToken) + { + if (timeout <= TimeSpan.Zero) + return; + + using var timeoutSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + timeoutSource.CancelAfter(timeout); + + try + { + await _wake.Reader.ReadAsync(timeoutSource.Token); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + // The fallback poll is due. + } + } + + /// + /// Polls one type for as many tasks as there are free worker slots, and starts each task it gets. + /// + private async Task PollAsync(WorkerQueue queue, CancellationToken cancellationToken) + { + // A slot is taken for every task the poll may return, so a polled task never waits for a + // worker. At least one, which is where the loop waits while every worker is busy. + await _slots.WaitAsync(cancellationToken); + var slots = 1; + while (slots < MaxBatchSize && _slots.Wait(0)) + slots++; + + var generation = queue.Generation; + + // TODO: Maybe this should be configurable, as in the other execution managers. + var workerId = Guid.NewGuid().ToString(); + + List tasks; + try + { + var polled = await _taskService.BatchPollAsync( + queue.Worker.TaskName, + workerId, + queue.Worker.TaskDomain ?? _configuration.Domain, + slots, + timeout: 0, + cancellationToken + ); + + tasks = polled?.Where(task => task != null).ToList() ?? new List(); + } + catch (ApiException exception) when (exception.StatusCode == 204) + { + tasks = new List(); + } + catch (Exception exception) when (!cancellationToken.IsCancellationRequested) + { + _slots.Release(slots); + _logger.LogError(exception, "Exception during the task polling"); + + // The type stays pending. Waiting first keeps a server that is down from being + // polled in a tight loop. + await Task.Delay(Math.Max(_configuration.SleepInterval, 1000), cancellationToken); + return; + } + catch + { + _slots.Release(slots); + throw; + } + + if (tasks.Count < slots) + { + _slots.Release(slots - tasks.Count); + queue.MarkDrained(generation); + } + + foreach (var task in tasks) + { + _ = RunAsync(task, workerId, queue.Worker, cancellationToken); + } + } + + private async Task RunAsync(Client.Generated.Task task, string workerId, TaskToWorker worker, CancellationToken cancellationToken) + { + try + { + await _runner.RunAsync(task, workerId, worker, cancellationToken); + } + catch (Exception exception) + { + // Nobody awaits this task, so an exception left here would go unseen. + _logger.LogError(exception, "Exception while running task {TaskId}", task.TaskId); + } + finally + { + _slots.Release(); + } + } + + /// + /// One registered worker's type, and whether tasks of it may be waiting. + /// + private sealed class WorkerQueue(TaskToWorker worker) + { + private readonly object _gate = new(); + private bool _pending; + private long _generation; + + public TaskToWorker Worker { get; } = worker; + + public bool IsPending + { + get + { + lock (_gate) + return _pending; + } + } + + /// + /// Counts the notices so far. A poll reads it before it starts and hands it to + /// . + /// + public long Generation + { + get + { + lock (_gate) + return _generation; + } + } + + public void MarkPending() + { + lock (_gate) + { + _generation++; + _pending = true; + } + } + + /// + /// Clears the type after a poll found its queue empty, unless a notice arrived after the + /// poll read . The comparison and the clear are one step, so + /// a notice cannot land between them and be lost. + /// + public void MarkDrained(long generation) + { + lock (_gate) + { + if (_generation == generation) + _pending = false; + } + } + } + } +} diff --git a/src/ConductorSharp.Engine/NotifiedPollingConfig.cs b/src/ConductorSharp.Engine/NotifiedPollingConfig.cs new file mode 100644 index 00000000..a9c9b8e2 --- /dev/null +++ b/src/ConductorSharp.Engine/NotifiedPollingConfig.cs @@ -0,0 +1,14 @@ +namespace ConductorSharp.Engine +{ + /// + /// Settings of the execution manager that polls on task availability notices. + /// + internal class NotifiedPollingConfig + { + /// + /// How often every task type is polled whether a notice came or not, in milliseconds. It + /// picks up tasks whose notice the transport lost. + /// + public int FallbackPollInterval { get; set; } + } +} diff --git a/src/ConductorSharp.Engine/PolledTaskRunner.cs b/src/ConductorSharp.Engine/PolledTaskRunner.cs new file mode 100644 index 00000000..84e62d88 --- /dev/null +++ b/src/ConductorSharp.Engine/PolledTaskRunner.cs @@ -0,0 +1,187 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using ConductorSharp.Client; +using ConductorSharp.Client.Generated; +using ConductorSharp.Client.Service; +using ConductorSharp.Client.Util; +using ConductorSharp.Engine.Exceptions; +using ConductorSharp.Engine.Interface; +using ConductorSharp.Engine.Model; +using ConductorSharp.Engine.Util; +using MediatR; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using Task = System.Threading.Tasks.Task; + +namespace ConductorSharp.Engine +{ + /// + /// Runs a polled task through its handler and reports the outcome to the server. Every execution + /// manager hands its polled tasks to this one. + /// + /// + /// Logs under the category, where this code lived before it was + /// shared, so existing log filters keep matching. + /// + internal class PolledTaskRunner( + ILogger logger, + ITaskService taskService, + IExternalPayloadService externalPayloadService, + IServiceScopeFactory lifetimeScope, + ICancellationNotifier cancellationNotifier + ) + { + private readonly ILogger _logger = logger; + private readonly ITaskService _taskManager = taskService; + private readonly IExternalPayloadService _externalPayloadService = externalPayloadService; + private readonly IServiceScopeFactory _lifetimeScopeFactory = lifetimeScope; + private readonly ICancellationNotifier _cancellationNotifier = cancellationNotifier; + + private static Type GetInputType(Type workerType) + { + var interfaces = workerType + .GetInterfaces() + .Where(a => a.IsGenericType && a.GetGenericTypeDefinition() == typeof(ITaskRequestHandler<,>)) + .First(); + var genericArguments = interfaces.GetGenericArguments(); + + var inputType = genericArguments[0]; + var outputType = genericArguments[1]; + + return inputType; + } + + public async Task RunAsync( + Client.Generated.Task pollResponse, + string workerId, + TaskToWorker scheduledWorker, + CancellationToken cancellationToken + ) + { + using var tokenHolder = _cancellationNotifier.GetCancellationToken(pollResponse.TaskId, cancellationToken); + + try + { + if (!string.IsNullOrEmpty(pollResponse.ExternalInputPayloadStoragePath)) + { + _logger.LogDebug("Fetching storage {location}", pollResponse.ExternalInputPayloadStoragePath); + // TODO: Check what the operation and payload type are + var externalStorageLocation = await _taskManager.GetExternalStorageLocationAsync( + pollResponse.ExternalInputPayloadStoragePath, + "READ", + "TASK_INPUT", + cancellationToken + ); + + // TODO: iffy + var file = await _externalPayloadService.GetExternalStorageDataAsync(externalStorageLocation.Path, tokenHolder.CancellationToken); + + using TextReader textReader = new StreamReader(file.Stream); + var json = await textReader.ReadToEndAsync(); + + pollResponse.InputData = JsonConvert.DeserializeObject>( + json, + ConductorConstants.IoJsonSerializerSettings + ); + } + + var inputType = GetInputType(scheduledWorker.TaskType); + var inputData = SerializationHelper.DictonaryToObject(inputType, pollResponse.InputData, ConductorConstants.IoJsonSerializerSettings); + // Poll response data can be huge (if read from external storage) + // We can save memory by not holding reference to pollResponse.InputData after it is parsed + pollResponse.InputData = null; + + using var scope = _lifetimeScopeFactory.CreateScope(); + + var context = scope.ServiceProvider.GetService(); + var mediator = scope.ServiceProvider.GetRequiredService(); + + if (context != null) + { + context.WorkflowName = pollResponse.WorkflowType; + context.TaskName = pollResponse.TaskDefName; + context.TaskReferenceName = pollResponse.ReferenceTaskName; + context.WorkflowId = pollResponse.WorkflowInstanceId; + context.CorrelationId = pollResponse.CorrelationId; + context.Context = ConductorSharpExecutionContext.CopyContext(pollResponse.Context); + context.TaskId = pollResponse.TaskId; + context.WorkerId = workerId; + } + + var response = await mediator.Send(inputData, tokenHolder.CancellationToken); + + await _taskManager.UpdateAsync( + new TaskResult + { + TaskId = pollResponse.TaskId, + Status = TaskResultStatus.COMPLETED, + OutputData = SerializationHelper.ObjectToDictionary(response, ConductorConstants.IoJsonSerializerSettings), + WorkflowInstanceId = pollResponse.WorkflowInstanceId + }, + tokenHolder.CancellationToken + ); + } + catch (OperationCanceledException) when (tokenHolder.IsCancellationRequestedByNotifier) + { + _logger.LogWarning( + "Polled task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) is cancelled", + pollResponse.TaskDefName, + pollResponse.TaskId, + pollResponse.WorkflowType, + pollResponse.WorkflowInstanceId + ); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) // This is fine since we know cancellationToken comes from background service + { + _logger.LogWarning( + "Cancelling task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) due to background service shutdown", + pollResponse.TaskDefName, + pollResponse.TaskId, + pollResponse.WorkflowType, + pollResponse.WorkflowInstanceId + ); + } + catch (Exception exception) + { + _logger.LogError( + "{@Exception} while executing {Task} as part of {Workflow} with id {WorkflowId}", + exception, + pollResponse.TaskDefName, + pollResponse.WorkflowType, + pollResponse.WorkflowInstanceId + ); + + // A worker may throw a StructuredErrorException to attach a sanitized, stable classification to the + // failed task's output. Plain exceptions keep producing only error_message, preserving backward compatibility. + var errorMessage = new ErrorOutput { ErrorMessage = exception.Message, StructuredError = StructuredError.FromException(exception) }; + + // TODO: We should verify that this is alright, it is possible that when executed concurrently, + // the updates caused by LogAsync will be discarded because the call of UpdateAsync(TaskResult...) + // sets the logs to null. Not sure how this is implemented in the backend, also, would have expected this to be a + // PUT or PATCH request, but by specs it is POST + await Task.WhenAll( + [ + _taskManager.UpdateAsync( + new TaskResult + { + TaskId = pollResponse.TaskId, + Status = TaskResultStatus.FAILED, + ReasonForIncompletion = exception.Message, + OutputData = SerializationHelper.ObjectToDictionary(errorMessage, ConductorConstants.IoJsonSerializerSettings), + WorkflowInstanceId = pollResponse?.WorkflowInstanceId + }, + tokenHolder.CancellationToken + ), + _taskManager.LogAsync(pollResponse.TaskId, exception.Message, tokenHolder.CancellationToken), + _taskManager.LogAsync(pollResponse.TaskId, exception.StackTrace, tokenHolder.CancellationToken) + ] + ); + } + } + } +} diff --git a/src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs b/src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs index f8d2e84e..cabb9b26 100644 --- a/src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs +++ b/src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs @@ -47,7 +47,9 @@ private async Task RunAsync(CancellationToken cancellationToken) await _healthService.SetExecutionManagerRunning(cancellationToken); await _executionManager.StartAsync(cancellationToken); } - catch (TaskCanceledException) + // A stop can surface from any wait the manager is in. Task.Delay throws + // TaskCanceledException, while a semaphore or a channel throws its base type. + catch (OperationCanceledException exception) when (exception is TaskCanceledException || cancellationToken.IsCancellationRequested) { _logger.LogInformation("Stopping ConductorSharp background service"); } diff --git a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs index 968044dd..8fdc52f0 100644 --- a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs +++ b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs @@ -28,25 +28,21 @@ internal class TypePollSpreadingExecutionManager : IExecutionManager private readonly WorkerSetConfig _configuration; private readonly ILogger _logger; private readonly ITaskService _taskManager; - private readonly IExternalPayloadService _externalPayloadService; private readonly IEnumerable _registeredWorkers; - private readonly IServiceScopeFactory _lifetimeScopeFactory; private readonly IPollTimingStrategy _pollTimingStrategy; private readonly IPollOrderStrategy _pollOrderStrategy; - private readonly ICancellationNotifier _cancellationNotifier; private readonly TaskQueuePollingService _taskQueuePollingService; + private readonly PolledTaskRunner _runner; public TypePollSpreadingExecutionManager( WorkerSetConfig options, ILogger logger, ITaskService taskService, IEnumerable workerMappings, - IExternalPayloadService externalPayloadService, - IServiceScopeFactory lifetimeScope, IPollTimingStrategy pollTimingStrategy, IPollOrderStrategy pollOrderStrategy, - ICancellationNotifier cancellationNotifier, - TaskQueuePollingService taskQueuePollingService + TaskQueuePollingService taskQueuePollingService, + PolledTaskRunner runner ) { _configuration = options; @@ -54,12 +50,10 @@ TaskQueuePollingService taskQueuePollingService _logger = logger; _taskManager = taskService; _registeredWorkers = workerMappings; - _lifetimeScopeFactory = lifetimeScope; _pollTimingStrategy = pollTimingStrategy; _pollOrderStrategy = pollOrderStrategy; - _cancellationNotifier = cancellationNotifier; - _externalPayloadService = externalPayloadService; _taskQueuePollingService = taskQueuePollingService; + _runner = runner; } public async Task StartAsync(CancellationToken cancellationToken) @@ -112,20 +106,6 @@ private string GetQueueTaskName(TaskToWorker taskToWorker) return taskToWorker.TaskName; } - private static Type GetInputType(Type workerType) - { - var interfaces = workerType - .GetInterfaces() - .Where(a => a.IsGenericType && a.GetGenericTypeDefinition() == typeof(ITaskRequestHandler<,>)) - .First(); - var genericArguments = interfaces.GetGenericArguments(); - - var inputType = genericArguments[0]; - var outputType = genericArguments[1]; - - return inputType; - } - private async Task PollAndHandle(TaskToWorker scheduledWorker, CancellationToken cancellationToken) { Client.Generated.Task pollResponse; @@ -160,135 +140,7 @@ private async Task PollAndHandle(TaskToWorker scheduledWorker, CancellationToken if (pollResponse == null) return; - await ProcessPolledTask(pollResponse, workerId, scheduledWorker, cancellationToken); - } - - private async Task ProcessPolledTask( - Client.Generated.Task pollResponse, - string workerId, - TaskToWorker scheduledWorker, - CancellationToken cancellationToken - ) - { - using var tokenHolder = _cancellationNotifier.GetCancellationToken(pollResponse.TaskId, cancellationToken); - - try - { - if (!string.IsNullOrEmpty(pollResponse.ExternalInputPayloadStoragePath)) - { - _logger.LogDebug("Fetching storage {location}", pollResponse.ExternalInputPayloadStoragePath); - // TODO: Check what the operation and payload type are - var externalStorageLocation = await _taskManager.GetExternalStorageLocationAsync( - pollResponse.ExternalInputPayloadStoragePath, - "READ", - "TASK_INPUT", - cancellationToken - ); - - // TODO: iffy - var file = await _externalPayloadService.GetExternalStorageDataAsync(externalStorageLocation.Path, tokenHolder.CancellationToken); - - using TextReader textReader = new StreamReader(file.Stream); - var json = await textReader.ReadToEndAsync(); - - pollResponse.InputData = JsonConvert.DeserializeObject>( - json, - ConductorConstants.IoJsonSerializerSettings - ); - } - - var inputType = GetInputType(scheduledWorker.TaskType); - var inputData = SerializationHelper.DictonaryToObject(inputType, pollResponse.InputData, ConductorConstants.IoJsonSerializerSettings); - // Poll response data can be huge (if read from external storage) - // We can save memory by not holding reference to pollResponse.InputData after it is parsed - pollResponse.InputData = null; - - using var scope = _lifetimeScopeFactory.CreateScope(); - - var context = scope.ServiceProvider.GetService(); - var mediator = scope.ServiceProvider.GetRequiredService(); - - if (context != null) - { - context.WorkflowName = pollResponse.WorkflowType; - context.TaskName = pollResponse.TaskDefName; - context.TaskReferenceName = pollResponse.ReferenceTaskName; - context.WorkflowId = pollResponse.WorkflowInstanceId; - context.CorrelationId = pollResponse.CorrelationId; - context.Context = ConductorSharpExecutionContext.CopyContext(pollResponse.Context); - context.TaskId = pollResponse.TaskId; - context.WorkerId = workerId; - } - - var response = await mediator.Send(inputData, tokenHolder.CancellationToken); - - await _taskManager.UpdateAsync( - new TaskResult - { - TaskId = pollResponse.TaskId, - Status = TaskResultStatus.COMPLETED, - OutputData = SerializationHelper.ObjectToDictionary(response, ConductorConstants.IoJsonSerializerSettings), - WorkflowInstanceId = pollResponse.WorkflowInstanceId - }, - tokenHolder.CancellationToken - ); - } - catch (OperationCanceledException) when (tokenHolder.IsCancellationRequestedByNotifier) - { - _logger.LogWarning( - "Polled task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) is cancelled", - pollResponse.TaskDefName, - pollResponse.TaskId, - pollResponse.WorkflowType, - pollResponse.WorkflowInstanceId - ); - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) // This is fine since we know cancellationToken comes from background service - { - _logger.LogWarning( - "Cancelling task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) due to background service shutdown", - pollResponse.TaskDefName, - pollResponse.TaskId, - pollResponse.WorkflowType, - pollResponse.WorkflowInstanceId - ); - } - catch (Exception exception) - { - _logger.LogError( - "{@Exception} while executing {Task} as part of {Workflow} with id {WorkflowId}", - exception, - pollResponse.TaskDefName, - pollResponse.WorkflowType, - pollResponse.WorkflowInstanceId - ); - - // A worker may throw a StructuredErrorException to attach a sanitized, stable classification to the - // failed task's output. Plain exceptions keep producing only error_message, preserving backward compatibility. - var errorMessage = new ErrorOutput { ErrorMessage = exception.Message, StructuredError = StructuredError.FromException(exception) }; - - // TODO: We should verify that this is alright, it is possible that when executed concurrently, - // the updates caused by LogAsync will be discarded because the call of UpdateAsync(TaskResult...) - // sets the logs to null. Not sure how this is implemented in the backend, also, would have expected this to be a - // PUT or PATCH request, but by specs it is POST - await Task.WhenAll( - [ - _taskManager.UpdateAsync( - new TaskResult - { - TaskId = pollResponse.TaskId, - Status = TaskResultStatus.FAILED, - ReasonForIncompletion = exception.Message, - OutputData = SerializationHelper.ObjectToDictionary(errorMessage, ConductorConstants.IoJsonSerializerSettings), - WorkflowInstanceId = pollResponse?.WorkflowInstanceId - }, - tokenHolder.CancellationToken - ), - _taskManager.LogAsync(pollResponse.TaskId, exception.Message, tokenHolder.CancellationToken), - _taskManager.LogAsync(pollResponse.TaskId, exception.StackTrace, tokenHolder.CancellationToken) - ] - ); - } + await _runner.RunAsync(pollResponse, workerId, scheduledWorker, cancellationToken); } } } diff --git a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj index a7c7a774..d9ef3bad 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 4.5.0 + 4.6.0 Codaxy Codaxy diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index f2273b42..eb27f339 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 4.5.0 + 4.6.0 diff --git a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj index 1d5c5a4e..e7c1fd35 100644 --- a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj +++ b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj @@ -7,7 +7,7 @@ disable true dotnet-conductorsharp - 4.5.0 + 4.6.0 diff --git a/test/ConductorSharp.Engine.Tests/Unit/NotifiedExecutionManagerTests.cs b/test/ConductorSharp.Engine.Tests/Unit/NotifiedExecutionManagerTests.cs new file mode 100644 index 00000000..b5f550f2 --- /dev/null +++ b/test/ConductorSharp.Engine.Tests/Unit/NotifiedExecutionManagerTests.cs @@ -0,0 +1,441 @@ +using System.Collections.Concurrent; +using ConductorSharp.Client.Generated; +using ConductorSharp.Client.Service; +using ConductorSharp.Engine.Builders.Metadata; +using ConductorSharp.Engine.Extensions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using ConductorTask = ConductorSharp.Client.Generated.Task; +using Task = System.Threading.Tasks.Task; + +namespace ConductorSharp.Engine.Tests.Unit +{ + /// + /// The execution manager behind UseTaskAvailabilityNotifications polls a type when the + /// transport announces it, drains its queue, and otherwise only polls on the fallback interval. + /// + public class NotifiedExecutionManagerTests + { + private const string TaskName = "NOTICE_probe"; + + /// + /// Long enough that no test sees a fallback poll unless it asks for one. + /// + private const int NoFallbackPoll = 600_000; + + [Fact] + public async Task Idle_worker_polls_once_on_start_and_then_only_on_a_notice() + { + var taskService = new QueueTaskService(); + var notifications = new FakeNotifications(); + await using var manager = await RunningManager.StartAsync(taskService, notifications, NoFallbackPoll); + + // The catch-up poll after subscribing, for tasks queued before the subscription. + await WaitUntilAsync(() => taskService.Polls == 1); + await Task.Delay(300); + Assert.Equal(1, taskService.Polls); + + taskService.Enqueue(ProbeTask("task-1")); + notifications.Notify(TaskName); + + await WaitUntilAsync(() => taskService.Updates.Count == 1); + Assert.Equal(TaskResultStatus.COMPLETED, taskService.Updates.Single().Status); + + // One task came back for the slots it asked for, so the queue was empty and the + // manager is idle again. + await Task.Delay(300); + Assert.Equal(2, taskService.Polls); + } + + [Fact] + public async Task A_notice_drains_the_queue_through_the_free_slots() + { + var taskService = new QueueTaskService(); + var notifications = new FakeNotifications(); + await using var manager = await RunningManager.StartAsync(taskService, notifications, NoFallbackPoll, maxConcurrentWorkers: 2); + await WaitUntilAsync(() => taskService.Polls == 1); + + for (var i = 0; i < 5; i++) + taskService.Enqueue(ProbeTask($"task-{i}", delayMs: 50)); + + notifications.Notify(TaskName); + + // One notice is enough: with two slots the manager polls again whenever a slot frees, + // until a poll comes back short. + await WaitUntilAsync(() => taskService.Updates.Count == 5); + Assert.True(taskService.LargestBatchRequested <= 2, $"Asked for {taskService.LargestBatchRequested} tasks with two slots"); + } + + [Fact] + public async Task A_notice_that_arrives_during_a_poll_is_not_lost() + { + var taskService = new QueueTaskService(); + var notifications = new FakeNotifications(); + + // The catch-up poll is held open, and answers empty however the queue looks by then: it + // read the queue before the task below was committed. + var heldPoll = taskService.HoldNextPoll(); + await using var manager = await RunningManager.StartAsync(taskService, notifications, NoFallbackPoll); + await WaitUntilAsync(() => taskService.Polls == 1); + + taskService.Enqueue(ProbeTask("task-1")); + notifications.Notify(TaskName); + heldPoll.SetResult(true); + + await WaitUntilAsync(() => taskService.Updates.Count == 1); + Assert.Equal(2, taskService.Polls); + } + + [Fact] + public async Task Fallback_poll_finds_a_task_whose_notice_was_lost() + { + var taskService = new QueueTaskService(); + var notifications = new FakeNotifications(); + await using var manager = await RunningManager.StartAsync(taskService, notifications, fallbackPollInterval: 200); + await WaitUntilAsync(() => taskService.Polls >= 1); + + taskService.Enqueue(ProbeTask("task-1")); + + await WaitUntilAsync(() => taskService.Updates.Count == 1); + } + + [Fact] + public async Task Subscribes_before_the_first_poll_and_unsubscribes_on_stop() + { + var taskService = new QueueTaskService(); + var notifications = new FakeNotifications(); + var manager = await RunningManager.StartAsync(taskService, notifications, NoFallbackPoll); + await WaitUntilAsync(() => taskService.Polls == 1); + + Assert.Equal(new[] { TaskName }, notifications.TaskTypes); + Assert.True(notifications.SubscribedAt <= taskService.FirstPollAt); + + await manager.DisposeAsync(); + + Assert.True(notifications.Unsubscribed); + } + + [Fact] + public async Task Retries_the_subscription_until_the_transport_accepts_it() + { + var taskService = new QueueTaskService(); + var notifications = new FakeNotifications { FailingSubscriptions = 1 }; + await using var manager = await RunningManager.StartAsync(taskService, notifications, NoFallbackPoll); + + await WaitUntilAsync(() => taskService.Polls == 1); + Assert.Equal(2, notifications.SubscribeAttempts); + } + + [Fact] + public void UseTaskAvailabilityNotifications_replaces_the_execution_manager() + { + var services = CreateServices(new QueueTaskService(), new FakeNotifications(), NoFallbackPoll, maxConcurrentWorkers: 1); + using var provider = services.BuildServiceProvider(); + + Assert.IsType(provider.GetRequiredService()); + } + + [Fact] + public void UseTaskAvailabilityNotifications_rejects_a_fallback_interval_that_is_not_positive() + { + var builder = new ServiceCollection() + .AddConductorSharp(baseUrl: "http://empty/empty") + .AddExecutionManager(1, 250, 500, null, typeof(NotifiedExecutionManagerTests).Assembly); + + Assert.Throws(() => builder.UseTaskAvailabilityNotifications(0)); + } + + private static ConductorTask ProbeTask(string taskId, int delayMs = 0) => + new() + { + TaskId = taskId, + TaskDefName = TaskName, + WorkflowInstanceId = "workflow-1", + InputData = new Dictionary { ["delay_ms"] = delayMs }, + }; + + private static async Task WaitUntilAsync(Func condition) + { + var deadline = DateTime.UtcNow.AddSeconds(10); + + while (!condition()) + { + Assert.True(DateTime.UtcNow < deadline, "The execution manager did not get there in time"); + await Task.Delay(10); + } + } + + private static ServiceCollection CreateServices( + QueueTaskService taskService, + FakeNotifications notifications, + int fallbackPollInterval, + int maxConcurrentWorkers + ) + { + var services = new ServiceCollection(); + services.AddLogging(); + services + .AddConductorSharp(baseUrl: "http://empty/empty") + .AddExecutionManager( + maxConcurrentWorkers: maxConcurrentWorkers, + sleepInterval: 250, + longPollInterval: 500, + domain: null, + handlerAssemblies: typeof(NotifiedExecutionManagerTests).Assembly + ) + .UseTaskAvailabilityNotifications(fallbackPollInterval); + services.RegisterWorkerTask(); + services.RemoveAll(); + services.AddSingleton(taskService); + services.AddSingleton(notifications); + + return services; + } + + /// + /// An execution manager running in the background, stopped on dispose. + /// + private sealed class RunningManager : IAsyncDisposable + { + private readonly ServiceProvider _provider; + private readonly CancellationTokenSource _stop = new(); + private Task _run; + + private RunningManager(ServiceProvider provider) + { + _provider = provider; + } + + public static Task StartAsync( + QueueTaskService taskService, + FakeNotifications notifications, + int fallbackPollInterval, + int maxConcurrentWorkers = 2 + ) + { + var provider = CreateServices(taskService, notifications, fallbackPollInterval, maxConcurrentWorkers).BuildServiceProvider(); + var running = new RunningManager(provider); + running._run = provider.GetRequiredService().StartAsync(running._stop.Token); + + return Task.FromResult(running); + } + + public async ValueTask DisposeAsync() + { + _stop.Cancel(); + await Assert.ThrowsAnyAsync(() => _run); + await _provider.DisposeAsync(); + } + } + + public class NoticeProbeRequest : IRequest + { + public int DelayMs { get; set; } + } + + public class NoticeProbeResponse { } + + [OriginalName(TaskName)] + public class NoticeProbeHandler : ITaskRequestHandler + { + public async Task Handle(NoticeProbeRequest request, CancellationToken cancellationToken) + { + if (request.DelayMs > 0) + await Task.Delay(request.DelayMs, cancellationToken); + + return new NoticeProbeResponse(); + } + } + + /// + /// Notices on demand, and a record of the subscription. + /// + private sealed class FakeNotifications : ITaskAvailabilityNotifications + { + private Action _onAvailable; + private int _subscribeAttempts; + + public int FailingSubscriptions { get; init; } + + public IReadOnlyCollection TaskTypes { get; private set; } + + public DateTime SubscribedAt { get; private set; } = DateTime.MaxValue; + + public bool Unsubscribed { get; private set; } + + public int SubscribeAttempts => _subscribeAttempts; + + public void Notify(string taskType) => _onAvailable?.Invoke(taskType); + + public Task SubscribeAsync( + IReadOnlyCollection taskTypes, + Action onAvailable, + CancellationToken cancellationToken = default + ) + { + if (Interlocked.Increment(ref _subscribeAttempts) <= FailingSubscriptions) + throw new InvalidOperationException("The transport is not connected"); + + TaskTypes = taskTypes; + _onAvailable = onAvailable; + SubscribedAt = DateTime.UtcNow; + + return Task.FromResult(new Subscription(this)); + } + + private sealed class Subscription(FakeNotifications owner) : IAsyncDisposable + { + public ValueTask DisposeAsync() + { + owner.Unsubscribed = true; + return ValueTask.CompletedTask; + } + } + } + + /// + /// A queue of probe tasks that batch polls take from. + /// + private sealed class QueueTaskService : ITaskService + { + private readonly ConcurrentQueue _queue = new(); + private TaskCompletionSource _heldPoll; + private int _polls; + private int _largestBatchRequested; + + public List Updates { get; } = []; + + public int Polls => _polls; + + public int LargestBatchRequested => _largestBatchRequested; + + public DateTime FirstPollAt { get; private set; } = DateTime.MaxValue; + + public void Enqueue(ConductorTask task) => _queue.Enqueue(task); + + /// + /// Holds the next poll open until the returned source is set, then answers it empty. + /// + public TaskCompletionSource HoldNextPoll() + { + _heldPoll = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + return _heldPoll; + } + + public async Task> BatchPollAsync( + string taskType, + string workerId, + string domain, + int? count = null, + int? timeout = null, + CancellationToken cancellationToken = default + ) + { + if (Interlocked.Increment(ref _polls) == 1) + FirstPollAt = DateTime.UtcNow; + + var requested = count ?? 1; + InterlockedMax(ref _largestBatchRequested, requested); + + var held = Interlocked.Exchange(ref _heldPoll, null); + if (held != null) + { + await held.Task; + return []; + } + + var tasks = new List(); + while (tasks.Count < requested && _queue.TryDequeue(out var task)) + tasks.Add(task); + + return tasks; + } + + public Task UpdateAsync(TaskResult updateRequest, CancellationToken cancellationToken = default) + { + lock (Updates) + Updates.Add(updateRequest); + + return Task.FromResult(updateRequest.TaskId); + } + + public Task LogAsync(string taskId, string message, CancellationToken cancellationToken = default) => Task.CompletedTask; + + private static void InterlockedMax(ref int target, int value) + { + int current; + while (value > (current = Volatile.Read(ref target))) + { + if (Interlocked.CompareExchange(ref target, value, current) == current) + return; + } + } + + public Task> ListQueuesAsync(CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task PollAsync(string taskType, string workerId, string domain, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task GetAsync(string taskId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + + public Task GetQueueAsync( + string taskType, + string domain = null, + string isolationGroupId = null, + string executionNamespace = null, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task GetExternalStorageLocationAsync( + string path, + string operation, + string payloadType, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task> ListLogsAsync(string taskId, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task> GetPollDataAsync(string taskType, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task> ListPollDataAsync(CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task>>> ListQueuesVerboseAsync( + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task RequeuePendingAsync(string taskType, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task SearchAsync( + int? start = null, + int? size = null, + string sort = null, + string freeText = null, + string query = null, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task SearchV2Async( + int? start = null, + int? size = null, + string sort = null, + string freeText = null, + string query = null, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task UpdateTaskByRefnameAsync( + string workflowId, + string taskRefName, + Status status, + string workerid, + IDictionary body, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + } + } +}