Distributed locks to prevent multiple pods executing Reload and Refresh (#7311)

* Distributed locks to prevent multiple pods executing Reload and Refresh

* Update src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionsRefresherDistributedLocking.cs

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* Update src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionsReloaderDistributedLocking.cs

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* Update src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionsRefresherDistributedLocking.cs

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* Moving distributed lock decorators to proper folder

* Adding logging for missing lock

* Fixing logger use in distributed classes

* Update src/modules/Elsa.Workflows.Runtime/Distributed/WorkflowDefinitionsReloaderDistributedLocking.cs

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* Update src/modules/Elsa.Workflows.Runtime/Distributed/WorkflowDefinitionsRefresherDistributedLocking.cs

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* Update src/modules/Elsa.Workflows.Runtime/Distributed/WorkflowDefinitionsRefresherDistributedLocking.cs

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* Update src/modules/Elsa.Workflows.Runtime/Distributed/WorkflowDefinitionsReloaderDistributedLocking.cs

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* Update src/modules/Elsa.Workflows.Runtime/Distributed/WorkflowDefinitionsRefresherDistributedLocking.cs

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* Update src/modules/Elsa.Workflows.Runtime/Distributed/WorkflowDefinitionsReloaderDistributedLocking.cs

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* Moving and renaming for clearer purpose

* Fixing build after file moves

* Addressing workflow refresher empty query leaking distributed lock logic

* Add status to WorkflowRefresherResponse model for in progress operation

* Removing typo, fixing build

* Update src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowDefinitionsRefresher.cs

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* Refactor `RefreshWorkflowDefinitionsAsync` for lock acquisition logic simplification.

---------

Co-authored-by: lucas.hipolito <lukhipolito@yahoo.com.br>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs

@@ -20,7 +20,14 @@ internal class Refresh(IWorkflowDefinitionsRefresher workflowDefinitionsRefreshe
     public override async Task HandleAsync(Request request, CancellationToken cancellationToken)
     {
         var result = await RefreshWorkflowDefinitionsAsync(request.DefinitionIds, cancellationToken);
-        await Send.OkAsync(new Response(result.Refreshed, result.NotFound), cancellationToken);
+        if (result.Status == RefreshWorkflowDefinitionsStatus.Completed)
+        {
+            await Send.OkAsync(new Response(result.Refreshed, result.NotFound), cancellationToken);
+        }
+        else
+        {
+            await Send.AcceptedAtAsync<Refresh>(responseBody: new Response(result.Refreshed, result.NotFound), cancellation: cancellationToken);
+        }
     }
 
     private async Task<RefreshWorkflowDefinitionsResponse> RefreshWorkflowDefinitionsAsync(ICollection<string>? definitionIds, CancellationToken cancellationToken)

src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs

@@ -29,6 +29,9 @@ public class DistributedRuntimeFeature(IModule module) : FeatureBase(module)
     {
         Services
             .AddScoped<DistributedWorkflowRuntime>()
-            .AddScoped<DistributedBookmarkQueueWorker>();
+            .AddScoped<DistributedBookmarkQueueWorker>()
+            
+            .Decorate<IWorkflowDefinitionsRefresher, DistributedWorkflowDefinitionsRefresher>()
+            .Decorate<IWorkflowDefinitionsReloader, DistributedWorkflowDefinitionsReloader>();
     }
 }

src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowDefinitionsRefresher.cs

@@ -0,0 +1,46 @@
+using Elsa.Workflows.Runtime.Requests;
+using Elsa.Workflows.Runtime.Responses;
+using JetBrains.Annotations;
+using Medallion.Threading;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Workflows.Runtime.Distributed;
+
+/// <summary>
+/// Decorator class that adds distributed locking to the Workflow Definitions Refresher.
+/// </summary>
+[UsedImplicitly]
+public class DistributedWorkflowDefinitionsRefresher(IWorkflowDefinitionsRefresher inner,
+    IDistributedLockProvider distributedLockProvider,
+    ILogger<DistributedWorkflowDefinitionsRefresher> logger) : IWorkflowDefinitionsRefresher
+{
+    /// <summary>
+    /// This ensures that only one instance of the application can refresh a set of workflow definitions at a time, preventing potential conflicts and ensuring consistency across distributed environments.
+    /// </summary>
+    public async Task<RefreshWorkflowDefinitionsResponse> RefreshWorkflowDefinitionsAsync(RefreshWorkflowDefinitionsRequest request, CancellationToken cancellationToken = default)
+    {
+        var isRefreshingAll = request.DefinitionIds == null || request.DefinitionIds.Count == 0;
+        var lockKey = isRefreshingAll
+            ? "WorkflowDefinitionsRefresher:All"
+            : $"WorkflowDefinitionsRefresher:{string.Join(",", request.DefinitionIds!.OrderBy(x => x))}";
+
+        await using var distributedLock = await distributedLockProvider.TryAcquireLockAsync(
+            lockKey,
+            TimeSpan.Zero,
+            cancellationToken);
+
+        if (distributedLock == null)
+        {
+            var logMessage = isRefreshingAll
+                ? "Could not acquire lock for refreshing all workflow definitions. Another instance is already refreshing all workflow definitions"
+                : "Could not acquire lock for refreshing workflow definitions. Another instance is already refreshing these workflow definitions";
+
+            logger.LogInformation(logMessage);
+
+            var failedDefinitionIds = isRefreshingAll ? Array.Empty<string>() : request.DefinitionIds!;
+            return new(Array.Empty<string>(), failedDefinitionIds, RefreshWorkflowDefinitionsStatus.AlreadyInProgress);
+        }
+
+        return await inner.RefreshWorkflowDefinitionsAsync(request, cancellationToken);
+    }
+}

src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowDefinitionsReloader.cs

@@ -0,0 +1,36 @@
+using JetBrains.Annotations;
+using Medallion.Threading;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Workflows.Runtime.Distributed;
+
+/// <summary>
+/// Decorator class that adds distributed locking to the Workflow Definitions Reloader.
+/// </summary>
+[UsedImplicitly]
+public class DistributedWorkflowDefinitionsReloader(
+    IWorkflowDefinitionsReloader inner,
+    IDistributedLockProvider distributedLockProvider,
+    ILogger<DistributedWorkflowDefinitionsReloader> logger) : IWorkflowDefinitionsReloader
+{
+    private const string LockKey = "WorkflowDefinitionsReloader";
+    
+    /// <summary>
+    /// This ensures that only one instance of the application can reload workflow definitions at a time, preventing potential conflicts and ensuring consistency across distributed environments.
+    /// </summary>
+    public async Task ReloadWorkflowDefinitionsAsync(CancellationToken cancellationToken = default)
+    {
+        await using var distributedLock = await distributedLockProvider.TryAcquireLockAsync(
+            LockKey,
+            TimeSpan.Zero,
+            cancellationToken);
+        
+        if (distributedLock == null)
+        {
+            logger.LogInformation("Could not acquire lock for workflow definitions reload. Another instance is already reloading workflow definitions");
+            return;
+        }
+        
+        await inner.ReloadWorkflowDefinitionsAsync(cancellationToken);
+    }
+}

src/modules/Elsa.Workflows.Runtime/Enums/RefreshWorkflowDefinitionsStatus.cs

@@ -0,0 +1,7 @@
+namespace Elsa.Workflows.Runtime;
+
+public enum RefreshWorkflowDefinitionsStatus
+{
+    Completed,
+    AlreadyInProgress
+}

src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs

@@ -3,4 +3,8 @@ namespace Elsa.Workflows.Runtime.Responses;
 /// <summary>
 /// Represents a response to a request to refresh workflow definitions.
 /// </summary>
-public record RefreshWorkflowDefinitionsResponse(ICollection<string> Refreshed, ICollection<string> NotFound);
+public record RefreshWorkflowDefinitionsResponse(
+    ICollection<string> Refreshed,
+    ICollection<string> NotFound,
+    RefreshWorkflowDefinitionsStatus Status = RefreshWorkflowDefinitionsStatus.Completed
+    );