From d89a07d391ab17d3ad9eeec60250221318e9cef1 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 21 May 2024 21:32:27 +0200 Subject: [PATCH] Add distributed locking to MassTransitWorkflowDispatcher (#5418) This commit introduces a distributed lock to the MassTransitWorkflowDispatcher to prevent concurrent updates to the workflow instance. The lock is acquired before any interaction with the workflow instance and is released after the update operation is complete. --- .../Services/MassTransitWorkflowDispatcher.cs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs index 9136ad4d2..93f50dfb6 100644 --- a/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs +++ b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs @@ -11,6 +11,7 @@ using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; using Elsa.Workflows.Runtime.Responses; using MassTransit; +using Medallion.Threading; using Microsoft.Extensions.Logging; namespace Elsa.MassTransit.Services; @@ -26,6 +27,7 @@ public class MassTransitWorkflowDispatcher( IBookmarkHasher bookmarkHasher, ITriggerStore triggerStore, IBookmarkStore bookmarkStore, + IDistributedLockProvider distributedLockProvider, ILogger logger) : IWorkflowDispatcher { @@ -163,6 +165,8 @@ public class MassTransitWorkflowDispatcher( if (input != null || properties != null) { + // Need to acquire a lock on the workflow instance to prevent concurrent updates. + await using var distributedLock = await distributedLockProvider.AcquireLockAsync(workflowInstanceId, TimeSpan.FromMinutes(2), cancellationToken); var workflowInstance = await workflowInstanceManager.FindByIdAsync(workflowInstanceId, cancellationToken); if (workflowInstance == null)