Poll on task availability notices (4.6.0)

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:

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

src/ConductorSharp.Client/ConductorSharp.Client.csproj

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

src/ConductorSharp.Client/Service/ITaskAvailabilityNotifications.cs

@@ -0,0 +1,28 @@
+namespace ConductorSharp.Client.Service
+{
+    /// <summary>
+    /// Tells workers when tasks of a type may be waiting, so they poll on notice instead of on a timer.
+    /// </summary>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    public interface ITaskAvailabilityNotifications
+    {
+        /// <summary>
+        /// Starts listening for notices about <paramref name="taskTypes"/>, and calls
+        /// <paramref name="onAvailable"/> with the task type of each one. Disposing the result stops
+        /// listening.
+        /// </summary>
+        /// <remarks>
+        /// 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.
+        /// </remarks>
+        Task<IAsyncDisposable> SubscribeAsync(
+            IReadOnlyCollection<string> taskTypes,
+            Action<string> onAvailable,
+            CancellationToken cancellationToken = default
+        );
+    }
+}

src/ConductorSharp.Engine/ConductorSharp.Engine.csproj

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

src/ConductorSharp.Engine/ExecutionManager.cs

@@ -28,38 +28,32 @@ internal class ExecutionManager : IExecutionManager
         private readonly WorkerSetConfig _configuration;
         private readonly ILogger<ExecutionManager> _logger;
         private readonly ITaskService _taskManager;
-        private readonly IExternalPayloadService _externalPayloadService;
         private readonly IEnumerable<TaskToWorker> _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<ExecutionManager> logger,
             ITaskService taskService,
             IEnumerable<TaskToWorker> workerMappings,
-            IExternalPayloadService externalPayloadService,
-            IServiceScopeFactory lifetimeScope,
             IPollTimingStrategy pollTimingStrategy,
             IPollOrderStrategy pollOrderStrategy,
-            ICancellationNotifier cancellationNotifier,
-            TaskQueuePollingService taskQueuePollingService
+            TaskQueuePollingService taskQueuePollingService,
+            PolledTaskRunner runner
         )
         {
             _configuration = options;
             _semaphore = new SemaphoreSlim(_configuration.MaxConcurrentWorkers);
             _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<IDictionary<string, object>>(
-                        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<ConductorSharpExecutionContext>();
-                var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
-
-                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);
         }
     }
 }

src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs

@@ -47,6 +47,8 @@ params Assembly[] handlerAssemblies
 
             Builder.AddSingleton<IExecutionManager, ExecutionManager>();
 
+            Builder.AddSingleton<PolledTaskRunner>();
+
             Builder.AddScoped<ConductorSharpExecutionContext>();
 
             Builder.AddTransient<FailedTaskStructuredErrorReader>();
@@ -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<IExecutionManager, NotifiedExecutionManager>();
+            return this;
+        }
+
         public IExecutionManagerBuilder AddPipelines(Action<IPipelineBuilder> behaviorBuilder)
         {
             var pipelineBuilder = new PipelineBuilder(Builder);

src/ConductorSharp.Engine/Extensions/IExecutionManagerBuilder.cs

@@ -13,5 +13,20 @@ IExecutionManagerBuilder SetHealthCheckService<T>()
             where T : IConductorSharpHealthService;
         IExecutionManagerBuilder UseConstantPollTimingStrategy();
         IExecutionManagerBuilder UseBetaExecutionManager();
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        /// <remarks>
+        /// Needs a transport that registers <see cref="Client.Service.ITaskAvailabilityNotifications"/>,
+        /// such as Maestro's NATS transport. Conductor's REST API has no such notices.
+        /// </remarks>
+        /// <param name="fallbackPollInterval">
+        /// 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.
+        /// </param>
+        IExecutionManagerBuilder UseTaskAvailabilityNotifications(int fallbackPollInterval = 30000);
     }
 }