feat: Enhance matching service with subscription and task dispatching capabilities; add system workers for cleanup, retries, and timers

Signed-off-by: Soeren Magnus Olesen <soeren@molesen.ch>

Author: df49b9cd

docs/PHASE1_PROGRESS.md

@@ -302,8 +302,8 @@ src/Odin.Persistence/
 ### 3. Execution Engine Services
 
 - **Objective**: Stand up History and Matching services that orchestrate shard ownership, task queues, and replay.
-- **Status**: 🧱 Infrastructure groundwork (schemas, shard repository, Go helpers) in place; service implementations pending.
-- **Immediate Focus**: Build HistoryService event pipelines, MatchingService task dispatch leveraging Hugo `TaskQueueChannelAdapter<T>`, and system workers (timer/retry/cleanup).
+- **Status**: ✅ History service persists/replays events; matching dispatch spins through `TaskQueueChannelAdapter<T>` with system workers running timers, retries, and cleanup over shared queues.
+- **Immediate Focus**: Broaden persistence coverage (timer/retry repositories) and integrate the services with Control Plane APIs.
 
 ### 4. Frontend & APIs
 
@@ -342,9 +342,9 @@ src/Odin.Persistence/
 
 ### Priority 3: Execution Engine
 
-- [ ] Build HistoryService with event persistence and replay
-- [ ] Build `MatchingService` with `TaskQueueChannelAdapter<T>`
-- [ ] Implement system workers (TimerWorker, RetryWorker, CleanupWorker)
+- [x] Build HistoryService with event persistence and replay
+- [x] Build `MatchingService` with `TaskQueueChannelAdapter<T>`
+- [x] Implement system workers (TimerWorker, RetryWorker, CleanupWorker)
 
 ### Priority 4: APIs
 

src/Odin.ExecutionEngine.Matching/MatchingService.cs

@@ -1,3 +1,4 @@
+using System.Text.Json;
 using System.Threading;
 using Hugo;
 using Microsoft.Extensions.Logging;
@@ -14,10 +15,12 @@ namespace Odin.ExecutionEngine.Matching;
 /// </summary>
 public sealed class MatchingService(
     ITaskQueueRepository taskQueueRepository,
-    ILogger<MatchingService> logger) : IMatchingService
+    ILogger<MatchingService> logger,
+    JsonSerializerOptions? serializerOptions = null) : IMatchingService
 {
     private readonly ITaskQueueRepository _taskQueueRepository = taskQueueRepository;
     private readonly ILogger<MatchingService> _logger = logger;
+    private readonly JsonSerializerOptions _serializerOptions = serializerOptions ?? JsonOptions.Default;
 
     /// <summary>
     /// Enqueues a workflow or activity task.
@@ -253,6 +256,32 @@ public async Task<Result<int>> ReclaimExpiredLeasesAsync(
                 Error.From($"Reclaim failed: {ex.Message}", OdinErrorCodes.TaskQueueError));
         }
     }
+
+    public Task<MatchingSubscription> SubscribeAsync(
+        string queueName,
+        string workerIdentity,
+        CancellationToken cancellationToken = default)
+    {
+        var dispatcher = new TaskQueueDispatcher(
+            queueName,
+            workerIdentity,
+            _taskQueueRepository,
+            _serializerOptions,
+            _logger);
+
+        if (cancellationToken.CanBeCanceled)
+        {
+#pragma warning disable CS4014
+            cancellationToken.Register(state =>
+            {
+                var d = (TaskQueueDispatcher)state!;
+                _ = d.DisposeAsync();
+            }, dispatcher, useSynchronizationContext: false);
+#pragma warning restore CS4014
+        }
+
+        return Task.FromResult(new MatchingSubscription(dispatcher));
+    }
 }
 
 /// <summary>
@@ -291,6 +320,11 @@ Task<Result<QueueStats>> GetQueueStatsAsync(
 
     Task<Result<int>> ReclaimExpiredLeasesAsync(
         CancellationToken cancellationToken = default);
+
+    Task<MatchingSubscription> SubscribeAsync(
+        string queueName,
+        string workerIdentity,
+        CancellationToken cancellationToken = default);
 }
 
 /// <summary>

src/Odin.ExecutionEngine.Matching/MatchingSubscription.cs

@@ -0,0 +1,17 @@
+using System.Threading.Channels;
+
+namespace Odin.ExecutionEngine.Matching;
+
+public sealed class MatchingSubscription : IAsyncDisposable
+{
+    private readonly TaskQueueDispatcher _dispatcher;
+
+    internal MatchingSubscription(TaskQueueDispatcher dispatcher)
+    {
+        _dispatcher = dispatcher;
+    }
+
+    public ChannelReader<MatchingTask> Reader => _dispatcher.Reader;
+
+    public ValueTask DisposeAsync() => _dispatcher.DisposeAsync();
+}

src/Odin.ExecutionEngine.Matching/MatchingTask.cs

@@ -0,0 +1,40 @@
+using Hugo;
+using Odin.Contracts;
+using Odin.Sdk;
+using static Hugo.Go;
+using Unit = Hugo.Go.Unit;
+
+namespace Odin.ExecutionEngine.Matching;
+
+public sealed class MatchingTask
+{
+    private readonly Func<CancellationToken, Task<Result<Unit>>> _complete;
+    private readonly Func<string, bool, CancellationToken, Task<Result<Unit>>> _fail;
+    private readonly Func<CancellationToken, Task<Result<TaskLease>>> _heartbeat;
+
+    internal MatchingTask(
+        TaskDispatchItem dispatchItem,
+        Func<CancellationToken, Task<Result<Unit>>> complete,
+        Func<string, bool, CancellationToken, Task<Result<Unit>>> fail,
+        Func<CancellationToken, Task<Result<TaskLease>>> heartbeat)
+    {
+        WorkflowTask = dispatchItem.WorkflowTask;
+        Lease = dispatchItem.Lease;
+        _complete = complete;
+        _fail = fail;
+        _heartbeat = heartbeat;
+    }
+
+    public WorkflowTask WorkflowTask { get; }
+
+    public TaskLease Lease { get; }
+
+    public Task<Result<Unit>> CompleteAsync(CancellationToken cancellationToken = default)
+        => _complete(cancellationToken);
+
+    public Task<Result<Unit>> FailAsync(string reason, bool requeue, CancellationToken cancellationToken = default)
+        => _fail(reason, requeue, cancellationToken);
+
+    public Task<Result<TaskLease>> HeartbeatAsync(CancellationToken cancellationToken = default)
+        => _heartbeat(cancellationToken);
+}

src/Odin.ExecutionEngine.Matching/Odin.ExecutionEngine.Matching.csproj

@@ -10,6 +10,7 @@
     <ProjectReference Include="../Odin.Contracts/Odin.Contracts.csproj" />
     <ProjectReference Include="../Odin.Core/Odin.Core.csproj" />
     <ProjectReference Include="../Odin.Persistence/Odin.Persistence.csproj" />
+    <ProjectReference Include="../Odin.Sdk/Odin.Sdk.csproj" />
   </ItemGroup>
 
 </Project>

src/Odin.ExecutionEngine.Matching/TaskDispatchItem.cs

@@ -0,0 +1,59 @@
+using System.Text.Json;
+using Hugo;
+using Microsoft.Extensions.Logging;
+using Odin.Contracts;
+using Odin.Core;
+using Odin.Persistence.Interfaces;
+using Odin.Sdk;
+using static Hugo.Go;
+
+namespace Odin.ExecutionEngine.Matching;
+
+internal sealed record TaskDispatchItem(
+    TaskLease Lease,
+    WorkflowTask WorkflowTask,
+    Func<CancellationToken, Task<Result<Unit>>> CompleteAsync,
+    Func<string, bool, CancellationToken, Task<Result<Unit>>> FailAsync,
+    Func<CancellationToken, Task<Result<TaskLease>>> HeartbeatAsync)
+{
+    public static Result<TaskDispatchItem> Create(
+        TaskLease lease,
+        ITaskQueueRepository repository,
+        JsonSerializerOptions serializerOptions,
+        ILogger logger)
+    {
+        WorkflowTask workflowTask;
+        try
+        {
+            workflowTask = DeserializeWorkflowTask(lease.Task, serializerOptions);
+        }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Failed to deserialize workflow task payload for lease {LeaseId}", lease.LeaseId);
+            return Result.Fail<TaskDispatchItem>(Error.FromException(ex));
+        }
+
+        async Task<Result<Unit>> CompleteAsync(CancellationToken cancellationToken)
+            => await repository.CompleteAsync(lease.LeaseId, cancellationToken).ConfigureAwait(false);
+
+        async Task<Result<Unit>> FailAsync(string reason, bool requeue, CancellationToken cancellationToken)
+            => await repository.FailAsync(lease.LeaseId, reason, requeue, cancellationToken).ConfigureAwait(false);
+
+        async Task<Result<TaskLease>> HeartbeatAsync(CancellationToken cancellationToken)
+            => await repository.HeartbeatAsync(lease.LeaseId, cancellationToken).ConfigureAwait(false);
+
+        return Result.Ok(new TaskDispatchItem(lease, workflowTask, CompleteAsync, FailAsync, HeartbeatAsync));
+    }
+
+    private static WorkflowTask DeserializeWorkflowTask(TaskQueueItem item, JsonSerializerOptions serializerOptions)
+    {
+        var json = item.TaskData.RootElement.GetRawText();
+        var workflowTask = JsonSerializer.Deserialize<WorkflowTask>(json, serializerOptions);
+        if (workflowTask is null)
+        {
+            throw new InvalidOperationException("Workflow task payload could not be deserialized.");
+        }
+
+        return workflowTask;
+    }
+}