Distributed locks to prevent multiple pods executing Reload and Refresh

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

@@ -0,0 +1,45 @@
+using Elsa.Workflows.Runtime.Requests;
+using Elsa.Workflows.Runtime.Responses;
+using JetBrains.Annotations;
+using Medallion.Threading;
+
+namespace Elsa.Workflows.Runtime.Distributed;
+
+/// <summary>
+/// Decorator class that adds Distributed Locking to the Workflow Definitions Refresher.
+/// </summary>
+/// <param name="inner"></param>
+/// <param name="distributedLockProvider"></param>
+[UsedImplicitly]
+public class WorkflowDefinitionsRefresherDistributedLocking(IWorkflowDefinitionsRefresher inner,
+    IDistributedLockProvider distributedLockProvider) : 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>
+    /// <param name="request"></param>
+    /// <param name="cancellationToken"></param>
+    /// <returns></returns>
+    public async Task<RefreshWorkflowDefinitionsResponse> RefreshWorkflowDefinitionsAsync(RefreshWorkflowDefinitionsRequest request, CancellationToken cancellationToken = default)
+    {
+        if (request.DefinitionIds == null 
+            || request.DefinitionIds.Count < 1)
+        {
+            return new RefreshWorkflowDefinitionsResponse(Array.Empty<string>(), Array.Empty<string>());
+        }
+        var definitionIdsKey = string.Join(",", request.DefinitionIds!.OrderBy(x => x));
+        var lockKey = $"WorkflowDefinitionsRefresher:{definitionIdsKey}";
+        await using var distributedLock = await distributedLockProvider.TryAcquireLockAsync(
+            lockKey,
+            TimeSpan.FromMinutes(1),
+            cancellationToken);
+
+        if (distributedLock == null)
+        {
+            // Another instance is already refreshing these workflow definitions. Skip this refresh.
+            return new RefreshWorkflowDefinitionsResponse(Array.Empty<string>(), request.DefinitionIds!);
+        }
+
+        return await inner.RefreshWorkflowDefinitionsAsync(request, cancellationToken);
+    }
+}

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

@@ -0,0 +1,37 @@
+using JetBrains.Annotations;
+using Medallion.Threading;
+
+namespace Elsa.Workflows.Runtime.Distributed;
+
+/// <summary>
+/// Decorator class that adds Distributed Locking to the Workflow Definitions Reloader.
+/// </summary>
+/// <param name="inner"></param>
+/// <param name="distributedLockProvider"></param>
+[UsedImplicitly]
+public class WorkflowDefinitionsReloaderDistributedLocking(
+    IWorkflowDefinitionsReloader inner,
+    IDistributedLockProvider distributedLockProvider) : 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>
+    /// <param name="cancellationToken"></param>
+    public async Task ReloadWorkflowDefinitionsAsync(CancellationToken cancellationToken = default)
+    {
+        await using var distributedLock = await distributedLockProvider.TryAcquireLockAsync(
+            LockKey,
+            TimeSpan.FromMinutes(1),
+            cancellationToken);
+
+        if (distributedLock == null)
+        {
+            // Another instance is already reloading workflow definitions. Skip this reload.
+            return;
+        }
+
+        await inner.ReloadWorkflowDefinitionsAsync(cancellationToken);
+    }
+}

src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs

@@ -13,6 +13,7 @@
 using Elsa.Workflows.Management.Contracts;
 using Elsa.Workflows.Management.Services;
 using Elsa.Workflows.Runtime.ActivationValidators;
+using Elsa.Workflows.Runtime.Distributed;
 using Elsa.Workflows.Runtime.Entities;
 using Elsa.Workflows.Runtime.Handlers;
 using Elsa.Workflows.Runtime.Options;
@@ -264,8 +265,15 @@ public override void Apply()
             .AddScoped<IActivityExecutionMapper, DefaultActivityExecutionMapper>()
             .AddScoped<IWorkflowDefinitionStorePopulator, DefaultWorkflowDefinitionStorePopulator>()
             .AddScoped<IRegistriesPopulator, DefaultRegistriesPopulator>()
+
             .AddScoped<IWorkflowDefinitionsRefresher, WorkflowDefinitionsRefresher>()
+            // Decorating with distributed locking to prevent multiple instances from refreshing the same workflow definitions at the same time.
+            .Decorate<IWorkflowDefinitionsRefresher, WorkflowDefinitionsRefresherDistributedLocking>()
+
             .AddScoped<IWorkflowDefinitionsReloader, WorkflowDefinitionsReloader>()
+            // Decorating with distributed locking to prevent multiple instances from reloading workflow definitions at the same time.
+            .Decorate<IWorkflowDefinitionsReloader, WorkflowDefinitionsReloaderDistributedLocking>()
+
             .AddScoped<IWorkflowRegistry, DefaultWorkflowRegistry>()
             .AddScoped<IWorkflowMatcher, WorkflowMatcher>()
             .AddScoped<IWorkflowInvoker, WorkflowInvoker>()