unique schedules deadletter override + servicprovider injection for error handler + executer externalized

Author: coronabytes

Core.TaskProcessor.SampleWebApi/Core.TaskProcessor.SampleWebApi.csproj

@@ -10,7 +10,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.2" />
     <PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.21.2" />
-    <PackageReference Include="Scalar.AspNetCore" Version="2.0.22" />
+    <PackageReference Include="Scalar.AspNetCore" Version="2.0.26" />
   </ItemGroup>
 
   <ItemGroup>

Core.TaskProcessor.SampleWebApi/Program.cs

@@ -7,21 +7,40 @@
 builder.Services.AddControllers();
 builder.Services.AddOpenApi();
 
-builder.Services.AddTaskProcessor(new TaskProcessorOptions
+builder.Services.AddTaskProcessor((sp, options) =>
 {
-    Redis = "localhost:6379,abortConnect=false",
-    Prefix = "{core}",
-    Queues = new[] { "high", "default" },
-    MaxWorkers = 4,
-    Retries = 3,
-    BaseFrequency = TimeSpan.FromSeconds(1),
-    Invisibility = TimeSpan.FromMinutes(5),
-    Retention = TimeSpan.FromDays(7),
-    UseHostedService = true,
-    UseCronSeconds = true
+    options.Redis = "localhost:6379,abortConnect=false";
+    options.Prefix = "{core}";
+    options.Queues = ["high", "default"];
+    options.MaxWorkers = 4;
+    options.Retries = 3;
+    options.Invisibility = TimeSpan.FromMinutes(5);
+    options.BaseFrequency = TimeSpan.FromSeconds(5);
+    options.PushbackFrequency = TimeSpan.FromSeconds(10);
+    options.CleanUpFrequency = TimeSpan.FromMinutes(5);
+    options.Retention = TimeSpan.FromDays(7);
+    options.Deadletter = true;
+    options.DeadletterUniqueSchedules = false;
+    options.UseCronSeconds = true;
+    options.OnTaskFailedDelay = (_, retry) =>
+        Task.FromResult(retry switch
+        {
+            2 => TimeSpan.FromSeconds(5),
+            1 => TimeSpan.FromSeconds(60),
+            _ => (TimeSpan?)null
+        });
+    options.OnTaskError = (_, exception) =>
+    {
+        sp.GetRequiredService<ILogger<ITaskProcessor>>()
+            .LogError(exception, "Task Error");
+
+        return Task.CompletedTask;
+    };
 });
+builder.Services.AddTaskProcessorExecutor();
 
 builder.Services.AddScoped<ISomeScopedService, SomeScopedService>();
+builder.Services.AddScoped<FaultyService>();
 
 var app = builder.Build();
 
@@ -37,9 +56,25 @@
 
 {
     var proc = app.Services.GetRequiredService<ITaskProcessor>();
+    await using var scope = app.Services.CreateAsyncScope();
+    var faulty = scope.ServiceProvider.GetRequiredService<FaultyService>();
+
+    //foreach (var schedule in await proc.GetSchedulesAsync("core", 0, 100))
+    //    await proc.CancelScheduleAsync(schedule.Id, "core");
+
+    await proc.ResumeAsync();
+
+    //var discarded = await proc.DiscardDeadTasksAsync("default");
 
-    foreach (var schedule in await proc.GetSchedulesAsync("core", 0, 100))
-        await proc.CancelScheduleAsync(schedule.Id, "core");
+    await proc.UpsertScheduleAsync(new ScheduleData
+    {
+        Id = "refresh",
+        Tenant = "core",
+        Queue = "default",
+        Cron = "0 */2 * * * *",
+        Timezone = "Etc/UTC",
+        Unique = true
+    }, () => faulty.DoFaultyStuff());
 }
 
 app.Run();

Core.TaskProcessor.SampleWebApi/Services/FaultyService.cs

@@ -0,0 +1,12 @@
+
+namespace Core.TaskProcessor.SampleWebApi.Services;
+
+public class FaultyService(ILogger<FaultyService> logger)
+{
+    public async Task DoFaultyStuff()
+    {
+        logger.LogInformation(nameof(DoFaultyStuff));
+        await Task.Delay(200);
+        throw new Exception("crash");
+    }
+}

Core.TaskProcessor/Core.TaskProcessor.csproj

@@ -23,7 +23,7 @@
 
 	<ItemGroup>
 		<PackageReference Include="Cronos" Version="0.9.0" />
-		<PackageReference Include="StackExchange.Redis" Version="2.8.24" />
+		<PackageReference Include="StackExchange.Redis" Version="2.8.31" />
 	</ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'net8.0' ">

Core.TaskProcessor/ServiceExtension.cs

@@ -6,11 +6,28 @@ public static class ServiceExtension
 {
     public static IServiceCollection AddTaskProcessor(this IServiceCollection collection, TaskProcessorOptions options)
     {
-        //collection.AddSingleton<IRemoteExpressionExecutor, RemoteExpressionExecutor>();
         collection.AddSingleton<ITaskProcessor>(new TaskProcessor(options));
 
-        if (options.UseHostedService)
-            collection.AddHostedService<TaskExecutorService>();
+        return collection;
+    }
+
+    public static IServiceCollection AddTaskProcessor(this IServiceCollection collection, Action<IServiceProvider, TaskProcessorOptions> configure)
+    {
+        var options = new TaskProcessorOptions();
+
+        collection.AddSingleton<ITaskProcessor>(sp =>
+        {
+            configure(sp, options);
+
+            return new TaskProcessor(options);
+        });
+
+        return collection;
+    }
+
+    public static IServiceCollection AddTaskProcessorExecutor(this IServiceCollection collection)
+    {
+        collection.AddHostedService<TaskExecutorService>();
 
         return collection;
     }

Core.TaskProcessor/TaskProcessor.cs

@@ -575,7 +575,7 @@ private async Task Process(TaskContext task)
                             tra.HashIncrementAsync(Prefix($"batch:{task.BatchId}"), "failed",
                                 flags: CommandFlags.FireAndForget);
                         //remaining = tra.HashDecrementAsync(Prefix($"batch:{task.BatchId}"), "remaining");
-                        if (_options.Deadletter)
+                        if (_options.Deadletter && (!string.IsNullOrEmpty(task.BatchId) || _options.DeadletterUniqueSchedules))
                             tra.SortedSetAddAsync(Prefix($"queue:{task.Queue}:deadletter"), task.TaskId,
                                 DateTimeOffset.UtcNow.ToUnixTimeSeconds());
                         else
@@ -663,7 +663,7 @@ private async Task Process(TaskContext task)
                         CommandFlags.FireAndForget);
                 }
 
-                if (!task.IsContinuation && string.IsNullOrEmpty(task.BatchId))
+                if (!task.IsContinuation && !string.IsNullOrEmpty(task.BatchId))
                     tra.HashIncrementAsync(Prefix($"batch:{task.BatchId}"), "duration",
                         (DateTimeOffset.UtcNow - start).TotalSeconds, CommandFlags.FireAndForget);
 #pragma warning restore CS4014

Core.TaskProcessor/TaskProcessorOptions.cs

@@ -49,6 +49,11 @@ public class TaskProcessorOptions
     /// </summary>
     public bool Deadletter { get; set; } = true;
 
+    /// <summary>
+    ///   when retries are exhausted on unique schedules they will never run again until task removed 
+    /// </summary>
+    public bool DeadletterUniqueSchedules { get; set; } = false;
+
     /// <summary>
     ///     deduplication window
     /// </summary>
@@ -60,7 +65,7 @@ public class TaskProcessorOptions
 
     public string Redis { get; set; } = string.Empty;
 
-    public bool UseHostedService { get; set; }
+    //public bool UseHostedService { get; set; }
     public bool UseCronSeconds { get; set; }
 
     public IRemoteExpressionExecutor ExpressionExecutor { get; set; } = new RemoteExpressionExecutor();