Implement Multi-Machine Orchestration Core

Author: Aitomates

src/GsdOrchestrator.Tests/GoalSchedulerTests.cs

@@ -8,13 +8,27 @@ public sealed class GoalSchedulerTests : IDisposable
 {
     private readonly string _root = Path.Combine(Path.GetTempPath(), $"scheduler-{Guid.NewGuid():N}");
 
+    private sealed class FakeNodeRegistry : INodeRegistry
+    {
+        public Task RegisterAsync(string nodeId, CancellationToken cancellationToken = default) => Task.CompletedTask;
+        public Task HeartbeatAsync(string nodeId, CancellationToken cancellationToken = default) => Task.CompletedTask;
+        public Task<IReadOnlyList<string>> GetActiveNodesAsync(CancellationToken cancellationToken = default) => Task.FromResult<IReadOnlyList<string>>(["worker-1"]);
+    }
+
+    private GoalScheduler CreateScheduler(IGoalStore store)
+    {
+        var nodeRegistry = new FakeNodeRegistry();
+        var options = Microsoft.Extensions.Options.Options.Create(new NodeSettings { NodeId = "worker-1" });
+        return new GoalScheduler(store, nodeRegistry, options, NullLogger<GoalScheduler>.Instance);
+    }
+
     [Fact]
     public async Task TryAcquireLease_IncompleteDependency_IsDenied()
     {
         var store = await CreateStore();
         var aggregate = AggregateWithTwoItems(WorkItemStatus.Pending);
         await store.SaveAsync(aggregate);
-        var scheduler = new GoalScheduler(store, NullLogger<GoalScheduler>.Instance);
+        var scheduler = CreateScheduler(store);
 
         var lease = await scheduler.TryAcquireLeaseAsync(Request("work-2"));
 
@@ -26,7 +40,7 @@ public async Task TryAcquireLease_CompetingCalls_RespectsGlobalLimitAtomically()
     {
         var store = await CreateStore();
         await store.SaveAsync(IndependentAggregate());
-        var scheduler = new GoalScheduler(store, NullLogger<GoalScheduler>.Instance);
+        var scheduler = CreateScheduler(store);
 
         var results = await Task.WhenAll(
             scheduler.TryAcquireLeaseAsync(Request("work-1", global: 1)),
@@ -42,7 +56,7 @@ public async Task TryAcquireLease_ProviderOrRepositoryLimit_DeniesSecond(int glo
     {
         var store = await CreateStore();
         await store.SaveAsync(IndependentAggregate());
-        var scheduler = new GoalScheduler(store, NullLogger<GoalScheduler>.Instance);
+        var scheduler = CreateScheduler(store);
         Assert.NotNull(await scheduler.TryAcquireLeaseAsync(Request("work-1", global, provider, repository)));
         Assert.Null(await scheduler.TryAcquireLeaseAsync(Request("work-2", global, provider, repository)));
     }
@@ -52,7 +66,7 @@ public async Task RecoverExpiredLeases_AllowsReacquire()
     {
         var store = await CreateStore();
         await store.SaveAsync(IndependentAggregate());
-        var scheduler = new GoalScheduler(store, NullLogger<GoalScheduler>.Instance);
+        var scheduler = CreateScheduler(store);
         var now = DateTimeOffset.UtcNow;
         Assert.NotNull(await scheduler.TryAcquireLeaseAsync(
             new LeaseRequest("goal-1", "work-1", "worker-1", now.AddMinutes(-2), now.AddMinutes(-1), 3, 3, 3)));

src/GsdOrchestrator/Scheduling/ApiGoalStore.cs

@@ -0,0 +1,64 @@
+using System.Net.Http.Json;
+using System.Text.Json;
+using Microsoft.Extensions.Logging;
+
+namespace GsdOrchestrator.Scheduling;
+
+public sealed class ApiGoalStore : IGoalStore
+{
+    private readonly HttpClient _httpClient;
+    private readonly ILogger<ApiGoalStore> _logger;
+
+    public ApiGoalStore(HttpClient httpClient, ILogger<ApiGoalStore> logger)
+    {
+        _httpClient = httpClient;
+        _logger = logger;
+    }
+
+    public Task InitializeAsync(CancellationToken cancellationToken = default)
+    {
+        _logger.LogInformation("ApiGoalStore initialized");
+        return Task.CompletedTask;
+    }
+
+    public async Task SaveAsync(GoalAggregate aggregate, CancellationToken cancellationToken = default)
+    {
+        var response = await _httpClient.PostAsJsonAsync($"/api/goals/{aggregate.Goal.Id}", aggregate, cancellationToken);
+        response.EnsureSuccessStatusCode();
+    }
+
+    public async Task<GoalAggregate?> LoadAsync(string goalId, CancellationToken cancellationToken = default)
+    {
+        var response = await _httpClient.GetAsync($"/api/goals/{goalId}", cancellationToken);
+        if (response.StatusCode == System.Net.HttpStatusCode.NotFound)
+            return null;
+        response.EnsureSuccessStatusCode();
+        return await response.Content.ReadFromJsonAsync<GoalAggregate>(cancellationToken: cancellationToken);
+    }
+
+    public async Task<LeaseRecord?> TryAcquireLeaseAsync(LeaseRequest request, CancellationToken cancellationToken = default)
+    {
+        var response = await _httpClient.PostAsJsonAsync($"/api/goals/{request.GoalId}/work-items/{request.WorkItemId}/leases", request, cancellationToken);
+        if (response.StatusCode == System.Net.HttpStatusCode.Conflict || response.StatusCode == System.Net.HttpStatusCode.NotFound)
+            return null;
+        response.EnsureSuccessStatusCode();
+        return await response.Content.ReadFromJsonAsync<LeaseRecord>(cancellationToken: cancellationToken);
+    }
+
+    public async Task<int> RecoverExpiredLeasesAsync(DateTimeOffset now, CancellationToken cancellationToken = default)
+    {
+        var response = await _httpClient.PostAsJsonAsync("/api/leases/recover", new { Now = now }, cancellationToken);
+        response.EnsureSuccessStatusCode();
+        return await response.Content.ReadFromJsonAsync<int>(cancellationToken: cancellationToken);
+    }
+
+    public async Task<bool> TryReserveIdempotencyKeyAsync(string goalId, string workItemId, string key, string effectType, CancellationToken cancellationToken = default)
+    {
+        var payload = new { GoalId = goalId, WorkItemId = workItemId, Key = key, EffectType = effectType };
+        var response = await _httpClient.PostAsJsonAsync("/api/idempotency-keys/reserve", payload, cancellationToken);
+        if (response.StatusCode == System.Net.HttpStatusCode.Conflict)
+            return false;
+        response.EnsureSuccessStatusCode();
+        return true;
+    }
+}

src/GsdOrchestrator/Scheduling/ApiNodeRegistry.cs

@@ -0,0 +1,37 @@
+using System.Net.Http.Json;
+using Microsoft.Extensions.Logging;
+
+namespace GsdOrchestrator.Scheduling;
+
+public sealed class ApiNodeRegistry : INodeRegistry
+{
+    private readonly HttpClient _httpClient;
+    private readonly ILogger<ApiNodeRegistry> _logger;
+
+    public ApiNodeRegistry(HttpClient httpClient, ILogger<ApiNodeRegistry> logger)
+    {
+        _httpClient = httpClient;
+        _logger = logger;
+    }
+
+    public async Task RegisterAsync(string nodeId, CancellationToken cancellationToken = default)
+    {
+        var response = await _httpClient.PostAsJsonAsync("/api/nodes/register", new { NodeId = nodeId }, cancellationToken);
+        response.EnsureSuccessStatusCode();
+        _logger.LogInformation("Node {NodeId} registered successfully.", nodeId);
+    }
+
+    public async Task HeartbeatAsync(string nodeId, CancellationToken cancellationToken = default)
+    {
+        var response = await _httpClient.PostAsJsonAsync("/api/nodes/heartbeat", new { NodeId = nodeId }, cancellationToken);
+        response.EnsureSuccessStatusCode();
+    }
+
+    public async Task<IReadOnlyList<string>> GetActiveNodesAsync(CancellationToken cancellationToken = default)
+    {
+        var response = await _httpClient.GetAsync("/api/nodes", cancellationToken);
+        response.EnsureSuccessStatusCode();
+        var nodes = await response.Content.ReadFromJsonAsync<string[]>(cancellationToken: cancellationToken);
+        return nodes ?? Array.Empty<string>();
+    }
+}

src/GsdOrchestrator/Scheduling/GoalScheduler.cs

@@ -1,25 +1,31 @@
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 
 namespace GsdOrchestrator.Scheduling;
 
 public sealed class GoalScheduler
 {
     private readonly IGoalStore _store;
+    private readonly INodeRegistry _nodeRegistry;
+    private readonly string _nodeId;
     private readonly ILogger<GoalScheduler> _logger;
 
-    public GoalScheduler(IGoalStore store, ILogger<GoalScheduler> logger)
+    public GoalScheduler(IGoalStore store, INodeRegistry nodeRegistry, IOptions<NodeSettings> options, ILogger<GoalScheduler> logger)
     {
         _store = store;
+        _nodeRegistry = nodeRegistry;
+        _nodeId = options.Value.NodeId;
         _logger = logger;
     }
 
     public async Task<LeaseRecord?> TryAcquireLeaseAsync(LeaseRequest request, CancellationToken cancellationToken = default)
     {
-        var lease = await _store.TryAcquireLeaseAsync(request, cancellationToken);
+        var nodeRequest = request with { Owner = _nodeId };
+        var lease = await _store.TryAcquireLeaseAsync(nodeRequest, cancellationToken);
         if (lease is null)
             _logger.LogDebug("Lease denied for goal {GoalId} work item {WorkItemId}", request.GoalId, request.WorkItemId);
         else
-            _logger.LogInformation("Lease {LeaseId} acquired for goal {GoalId} work item {WorkItemId}", lease.Id, request.GoalId, request.WorkItemId);
+            _logger.LogInformation("Lease {LeaseId} acquired for goal {GoalId} work item {WorkItemId} by node {NodeId}", lease.Id, request.GoalId, request.WorkItemId, _nodeId);
         return lease;
     }
 

src/GsdOrchestrator/Scheduling/INodeRegistry.cs

@@ -0,0 +1,8 @@
+namespace GsdOrchestrator.Scheduling;
+
+public interface INodeRegistry
+{
+    Task RegisterAsync(string nodeId, CancellationToken cancellationToken = default);
+    Task HeartbeatAsync(string nodeId, CancellationToken cancellationToken = default);
+    Task<IReadOnlyList<string>> GetActiveNodesAsync(CancellationToken cancellationToken = default);
+}

src/GsdOrchestrator/Scheduling/NodeSettings.cs

@@ -0,0 +1,6 @@
+namespace GsdOrchestrator.Scheduling;
+
+public sealed class NodeSettings
+{
+    public string NodeId { get; set; } = Environment.MachineName;
+}