non batch tasks + valkey test + redisinsight update

coronabytes · coronabytes · commit 8c181fd2ee8b · 2025-03-15T11:46:18.000+01:00
diff --git a/Core.TaskProcessor.SampleWebApi/Core.TaskProcessor.SampleWebApi.csproj b/Core.TaskProcessor.SampleWebApi/Core.TaskProcessor.SampleWebApi.csproj
@@ -8,9 +8,9 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.2" />
+    <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.3" />
     <PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.21.2" />
-    <PackageReference Include="Scalar.AspNetCore" Version="2.0.26" />
+    <PackageReference Include="Scalar.AspNetCore" Version="2.0.34" />
   </ItemGroup>
 
   <ItemGroup>
diff --git a/Core.TaskProcessor.SampleWebApi/Program.cs b/Core.TaskProcessor.SampleWebApi/Program.cs
@@ -9,7 +9,7 @@
 
 builder.Services.AddTaskProcessor((sp, options) =>
 {
-    options.Redis = "localhost:6379,abortConnect=false";
+    options.Redis = "localhost:6380,abortConnect=false";
     options.Prefix = "{core}";
     options.Queues = ["high", "default"];
     options.MaxWorkers = 4;
@@ -20,19 +20,22 @@
     options.CleanUpFrequency = TimeSpan.FromMinutes(5);
     options.Retention = TimeSpan.FromDays(7);
     options.Deadletter = true;
-    options.DeadletterUniqueSchedules = false;
+    options.DeadletterSchedules = true;
     options.UseCronSeconds = true;
     options.OnTaskFailedDelay = (_, retry) =>
         Task.FromResult(retry switch
         {
             2 => TimeSpan.FromSeconds(5),
-            1 => TimeSpan.FromSeconds(60),
+            1 => TimeSpan.FromSeconds(10),
             _ => (TimeSpan?)null
         });
     options.OnTaskError = (_, exception) =>
     {
+        //sp.GetRequiredService<ILogger<ITaskProcessor>>()
+        //    .LogError(exception, "Task Error");
+
         sp.GetRequiredService<ILogger<ITaskProcessor>>()
-            .LogError(exception, "Task Error");
+            .LogError(exception.Message);
 
         return Task.CompletedTask;
     };
@@ -58,23 +61,28 @@
     var proc = app.Services.GetRequiredService<ITaskProcessor>();
     await using var scope = app.Services.CreateAsyncScope();
     var faulty = scope.ServiceProvider.GetRequiredService<FaultyService>();
+    var svc = scope.ServiceProvider.GetRequiredService<ISomeScopedService>();
 
-    //foreach (var schedule in await proc.GetSchedulesAsync("core", 0, 100))
-    //    await proc.CancelScheduleAsync(schedule.Id, "core");
+    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");
 
-    await proc.UpsertScheduleAsync(new ScheduleData
-    {
-        Id = "refresh",
-        Tenant = "core",
-        Queue = "default",
-        Cron = "0 */2 * * * *",
-        Timezone = "Etc/UTC",
-        Unique = true
-    }, () => faulty.DoFaultyStuff());
+    //await proc.UpsertScheduleAsync(new ScheduleData
+    //{
+    //    Id = "refresh",
+    //    Tenant = "core",
+    //    Queue = "default",
+    //    Cron = "0 */2 * * * *",
+    //    Timezone = "Etc/UTC",
+    //    Unique = true
+    //}, () => faulty.DoFaultyStuff());
+
+    await proc.EnqueueTaskAsync("default", "core", () => faulty.DoFaultyStuff());
+    //await proc.EnqueueTaskAsync("default", "core", () => svc.DoSomethingAsync("test", CancellationToken.None));
+    //await proc.EnqueueTaskAsync("default", "core", () => svc.DoSomethingAsync("test-delay", CancellationToken.None), delayUntil: DateTimeOffset.UtcNow.AddSeconds(5));
 }
 
 app.Run();
diff --git a/Core.TaskProcessor/Core.TaskProcessor.csproj b/Core.TaskProcessor/Core.TaskProcessor.csproj
@@ -32,8 +32,8 @@
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'net9.0' ">
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.2" />
-    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.2" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.3" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.3" />
   </ItemGroup>
 
 </Project>
diff --git a/Core.TaskProcessor/ITaskProcessor.cs b/Core.TaskProcessor/ITaskProcessor.cs
@@ -35,6 +35,23 @@ public interface ITaskProcessor
     Task<string> EnqueueBatchAsync(string queue, string tenant, List<TaskData> tasks,
         List<TaskData>? continuations = null, string? scope = null);
 
+    /// <summary>
+    ///   enqueue single task without batch
+    /// </summary>
+    /// <param name="queue">queue to run tasks and continuations on</param>
+    /// <param name="tenant">tenant id</param>
+    /// <param name="task"></param>
+    /// <returns>task id</returns>
+    Task<string> EnqueueTaskAsync(string queue, string tenant, TaskData task);
+
+    /// <summary>
+    ///   enqueue single task without batch
+    /// </summary>
+    /// <param name="queue">queue to run tasks and continuations on</param>
+    /// <param name="tenant">tenant id</param>
+    /// <param name="methodCall"></param>
+    /// <returns>task id</returns>
+    Task<string> EnqueueTaskAsync(string queue, string tenant, Expression<Func<Task>> methodCall, DateTimeOffset? delayUntil = null, int? retries = null);
 
     /// <summary>
     ///     appends tasks to existing a batch. continuations will run again if batch previously completed.
diff --git a/Core.TaskProcessor/TaskProcessor.cs b/Core.TaskProcessor/TaskProcessor.cs
@@ -139,6 +139,52 @@ public async Task<string> EnqueueBatchAsync(string queue, string tenant, List<Ta
         return batchId;
     }
 
+    public async Task<string> EnqueueTaskAsync(string queue, string tenant, TaskData task)
+    {
+        var taskId = Guid.NewGuid().ToString("D");
+
+        var q = task.Queue ?? queue;
+
+        var db = _redis.GetDatabase();
+        var tra = db.CreateTransaction();
+#pragma warning disable CS4014
+        
+        tra.HashSetAsync(Prefix($"task:{taskId}"), [
+            new HashEntry("id", taskId),
+            new HashEntry("tenant", tenant),
+            new HashEntry("data", task.Data),
+            new HashEntry("topic", task.Topic),
+            new HashEntry("queue", q),
+            new HashEntry("retries", task.Retries ?? _options.Retries)
+        ]);
+
+        if (task.DelayUntil.HasValue)
+            tra.SortedSetAddAsync(Prefix($"queue:{queue}:pushback"), taskId,
+                task.DelayUntil.Value.ToUnixTimeSeconds());
+        else
+        {
+            tra.SortedSetAddAsync(Prefix("queues"), q, DateTimeOffset.UtcNow.ToUnixTimeSeconds());
+            tra.ListLeftPushAsync(Prefix($"queue:{q}"), taskId);
+            tra.PublishAsync(RedisChannel.Literal(Prefix($"queue:{q}:event")), "fetch");
+        }
+
+#pragma warning enable CS4014
+            await tra.ExecuteAsync().ConfigureAwait(false);
+        return taskId;
+    }
+
+    public Task<string> EnqueueTaskAsync(string queue, string tenant, Expression<Func<Task>> methodCall, DateTimeOffset? delayUntil = null, int? retries = null)
+    {
+        return EnqueueTaskAsync(queue, tenant, new TaskData
+        {
+            Topic = "internal:expression:v1",
+            Data = _options.ExpressionExecutor.Serialize(methodCall),
+            DelayUntil = delayUntil,
+            Queue = queue,
+            Retries = retries
+        });
+    }
+
     public async Task<bool> AppendBatchAsync(string queue, string tenant, string batchId, List<TaskData> tasks)
     {
         var db = _redis.GetDatabase();
@@ -306,6 +352,29 @@ public async Task<bool> FetchAsync()
                 _tasks.TryAdd(info.TaskId, info);
                 _actionBlock.Post(info);
             }
+            else // single task
+            {
+                var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(_shutdown.Token);
+
+                var info = new TaskContext
+                {
+                    Processor = this,
+                    TaskId = j,
+                    Tenant = (string)taskData["tenant"]!,
+                    Queue = (string?)taskData["queue"],
+                    CancelSource = linkedCts,
+                    CancelToken = linkedCts.Token,
+                    Topic = (string?)taskData["topic"] ?? string.Empty,
+                    Data = (byte[])taskData["data"]!,
+                    BatchId = null,
+                    Retries = (int?)taskData["retries"],
+                    IsCancellation = false,
+                    IsContinuation = false
+                };
+
+                _tasks.TryAdd(info.TaskId, info);
+                _actionBlock.Post(info);
+            }
 
             return true;
         }
@@ -575,7 +644,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 && (!string.IsNullOrEmpty(task.BatchId) || _options.DeadletterUniqueSchedules))
+                        if (_options.Deadletter && (string.IsNullOrEmpty(task.ScheduleId) || _options.DeadletterSchedules))
                             tra.SortedSetAddAsync(Prefix($"queue:{task.Queue}:deadletter"), task.TaskId,
                                 DateTimeOffset.UtcNow.ToUnixTimeSeconds());
                         else
diff --git a/Core.TaskProcessor/TaskProcessorOptions.cs b/Core.TaskProcessor/TaskProcessorOptions.cs
@@ -52,7 +52,7 @@ public class TaskProcessorOptions
     /// <summary>
     ///   when retries are exhausted on unique schedules they will never run again until task removed 
     /// </summary>
-    public bool DeadletterUniqueSchedules { get; set; } = false;
+    public bool DeadletterSchedules { get; set; } = false;
 
     /// <summary>
     ///     deduplication window
diff --git a/redis/docker-compose.yml b/redis/docker-compose.yml
@@ -1,5 +1,3 @@
-version: '3.9'
-
 services:
 
   redis:
@@ -12,17 +10,29 @@ services:
       - redis:/data
     command: redis-server --appendonly yes
     
+  valkey:
+    container_name: valkey
+    image: valkey/valkey:8.1-alpine
+    restart: unless-stopped
+    ports:
+      - 6380:6379
+    volumes:
+      - valkey:/data
+    command: valkey-server --appendonly yes
+    
   redisinsight:
     container_name: redisinsight
-    image: redislabs/redisinsight:latest
+    image: redis/redisinsight:latest
     restart: unless-stopped
     volumes:
-      - redisinsight:/db
+      - redisinsight:/data
     depends_on:
       - redis
+      - valkey
     ports:
-      - 8001:8001
+      - 5540:5540
 
 volumes:
   redis:
+  valkey:
   redisinsight:
