SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Acquire lock on correlation ID when dispatching new workflows

Browse source 
This fixes a race condition between initial workflow execution that e.g. sends a message to a topic, which is then handled by the same workflow later on.
But if that message is delivered too soon (before the workflow got persisted & indexed), then the message consumer won't be able to find the workflow.

Acquiring a lock on the correlation ID fixes this by synchronizing access to that resource
This commit is contained in:
Sipke Schoorstra 2021-06-03 14:02:07 +02:00
parent 44d756241a
commit 97df3b7bc7
7 changed files with 81 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/TopicMessageReceivedBookmark.cs
View file

@ -11,16 +11,14 @@ namespace Elsa.Activities.AzureServiceBus.Bookmarks
{
}
public TopicMessageReceivedBookmark(string topicName, string subscriptionName, string? correlationId = default)
public TopicMessageReceivedBookmark(string topicName, string subscriptionName)
{
TopicName = topicName;
SubscriptionName = subscriptionName;
CorrelationId = correlationId;
}
public string TopicName { get; set; } = default!;
public string TopicName { get; set; } = default!;
public string SubscriptionName { get; set; } = default!;
public string? CorrelationId { get; set; }
}
public class TopicMessageReceivedBookmarkProvider : BookmarkProvider<TopicMessageReceivedBookmark, AzureServiceBusTopicMessageReceived>
@ -32,7 +30,6 @@ namespace Elsa.Activities.AzureServiceBus.Bookmarks
{
TopicName = (await context.ReadActivityPropertyAsync(x => x.TopicName, cancellationToken))!,
SubscriptionName = (await context.ReadActivityPropertyAsync(x => x.SubscriptionName, cancellationToken))!,
CorrelationId = context.ActivityExecutionContext.WorkflowExecutionContext.CorrelationId
})
};
}

2
src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs
View file

@ -6,6 +6,7 @@ using Elsa.Activities.AzureServiceBus.Services;
using Elsa.Activities.AzureServiceBus.StartupTasks;
using Elsa.Events;
using Elsa.Runtime;
using Elsa.Services;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Management;
using Microsoft.Extensions.DependencyInjection;
@ -32,6 +33,7 @@ namespace Elsa.Activities.AzureServiceBus.Extensions
.AddSingleton<ITopicMessageReceiverFactory>(sp => sp.GetRequiredService<BusClientFactory>())
.AddSingleton<IServiceBusQueuesStarter, ServiceBusQueuesStarter>()
.AddSingleton<IServiceBusTopicsStarter, ServiceBusTopicsStarter>()
.AddSingleton<Scoped<IWorkflowLaunchpad>>()
.AddStartupTask<StartServiceBusQueues>()
.AddStartupTask<StartServiceBusTopics>()
.AddBookmarkProvider<QueueMessageReceivedBookmarkProvider>()

5
src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs
View file

@ -4,6 +4,7 @@ using Elsa.Activities.AzureServiceBus.Bookmarks;
using Elsa.Activities.AzureServiceBus.Options;
using Elsa.Bookmarks;
using Elsa.Dispatch;
using Elsa.Services;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Core;
using Microsoft.Extensions.Logging;
@ -15,10 +16,10 @@ namespace Elsa.Activities.AzureServiceBus.Services
{
public QueueWorker(
IReceiverClient messageReceiver,
IWorkflowDispatcher workflowDispatcher,
Scoped<IWorkflowLaunchpad> workflowLaunchpad,
IOptions<AzureServiceBusOptions> options,
Func<IReceiverClient, Task> disposeReceiverAction,
ILogger<QueueWorker> logger) : base(messageReceiver, workflowDispatcher, options, disposeReceiverAction, logger)
ILogger<QueueWorker> logger) : base(messageReceiver, workflowLaunchpad, options, disposeReceiverAction, logger)
{
}

7
src/activities/Elsa.Activities.AzureServiceBus/Services/TopicWorker.cs
View file

@ -4,6 +4,7 @@ using Elsa.Activities.AzureServiceBus.Bookmarks;
using Elsa.Activities.AzureServiceBus.Options;
using Elsa.Bookmarks;
using Elsa.Dispatch;
using Elsa.Services;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Core;
using Microsoft.Extensions.Logging;
@ -15,10 +16,10 @@ namespace Elsa.Activities.AzureServiceBus.Services
{
public TopicWorker(
IReceiverClient receiverClient,
IWorkflowDispatcher workflowDispatcher,
Scoped<IWorkflowLaunchpad> workflowLaunchpad,
IOptions<AzureServiceBusOptions> options,
Func<IReceiverClient, Task> disposeReceiverAction,
ILogger<TopicWorker> logger) : base(receiverClient, workflowDispatcher, options, disposeReceiverAction, logger)
ILogger<TopicWorker> logger) : base(receiverClient, workflowLaunchpad, options, disposeReceiverAction, logger)
{
}
@ -27,7 +28,7 @@ namespace Elsa.Activities.AzureServiceBus.Services
protected override IBookmark CreateBookmark(Message message)
{
GetTopicAndSubscription(out var topicName, out var subscriptionName);
return new TopicMessageReceivedBookmark(topicName, subscriptionName, message.CorrelationId);
return new TopicMessageReceivedBookmark(topicName, subscriptionName);
}
protected override IBookmark CreateTrigger(Message message)

13
src/activities/Elsa.Activities.AzureServiceBus/Services/WorkerBase.cs
View file

@ -6,6 +6,7 @@ using Elsa.Activities.AzureServiceBus.Models;
using Elsa.Activities.AzureServiceBus.Options;
using Elsa.Bookmarks;
using Elsa.Dispatch;
using Elsa.Services;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Core;
using Microsoft.Extensions.Logging;
@ -16,21 +17,21 @@ namespace Elsa.Activities.AzureServiceBus.Services
public abstract class WorkerBase : IAsyncDisposable
{
// TODO: Design multi-tenancy.
private const string TenantId = default;
private const string? TenantId = default;
private readonly IWorkflowDispatcher _workflowDispatcher;
private readonly Scoped<IWorkflowLaunchpad> _workflowLaunchpad;
private readonly Func<IReceiverClient, Task> _disposeReceiverAction;
private readonly ILogger _logger;
protected WorkerBase(
IReceiverClient receiverClient,
IWorkflowDispatcher workflowDispatcher,
Scoped<IWorkflowLaunchpad> workflowLaunchpad,
IOptions<AzureServiceBusOptions> options,
Func<IReceiverClient, Task> disposeReceiverAction,
ILogger logger)
{
ReceiverClient = receiverClient;
_workflowDispatcher = workflowDispatcher;
_workflowLaunchpad = workflowLaunchpad;
_disposeReceiverAction = disposeReceiverAction;
_logger = logger;
@ -73,7 +74,9 @@ namespace Elsa.Activities.AzureServiceBus.Services
var bookmark = CreateBookmark(message);
var trigger = CreateTrigger(message);
await _workflowDispatcher.DispatchAsync(new TriggerWorkflowsRequest(ActivityType, bookmark, trigger, model, correlationId, TenantId: TenantId), cancellationToken);
var launchContext = new CollectWorkflowsContext(ActivityType, bookmark, trigger, correlationId);
await _workflowLaunchpad.UseServiceAsync(service => service.CollectAndDispatchWorkflowsAsync(launchContext, model, cancellationToken));
}
private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e)

12
src/core/Elsa.Core/Builders/InlineActivityBuilderExtensions.cs
View file

@ -27,6 +27,9 @@ namespace Elsa.Builders
public static IActivityBuilder Then(this IBuilder builder, Action activity) =>
builder.Then<Inline>(inline => inline.Set(x => x.Function, RunInline(activity)));
public static IActivityBuilder Then(this IBuilder builder, Func<Task> activity) =>
builder.Then<Inline>(inline => inline.Set(x => x.Function, RunInline(activity)));
private static Func<ActivityExecutionContext, ValueTask<IActivityExecutionResult>> RunInline(
Func<ActivityExecutionContext, ValueTask<IActivityExecutionResult>> activity) =>
@ -60,7 +63,14 @@ namespace Elsa.Builders
context =>
{
activity();
return new ValueTask<IActivityExecutionResult>(new OutcomeResult());
return new ValueTask<IActivityExecutionResult>(new DoneResult());
};
private static Func<ActivityExecutionContext, ValueTask<IActivityExecutionResult>> RunInline(Func<Task> activity) =>
async context =>
{
await activity();
return new DoneResult();
};
}
}

113
src/core/Elsa.Core/Services/WorkflowLaunchpad.cs
View file

@ -1,4 +1,5 @@
using System.Collections.Generic;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
@ -110,59 +111,41 @@ namespace Elsa.Services
if (!ValidatePreconditions(workflowDefinitionId, workflowBlueprint))
return null;
var correlationLockHandle = default(IDistributedSynchronizationHandle?);
correlationId ??= Guid.NewGuid().ToString("N");
// If we are creating a correlated workflow, make sure to acquire a lock on it to prevent duplicate workflow instances from being created.
if (!string.IsNullOrWhiteSpace(correlationId))
// Acquire a lock on correlation ID to prevent duplicate workflow instances from being created.
await using var correlationLockHandle = await AcquireLockAsync(correlationId, cancellationToken);
// Acquire a lock on the workflow definition so that we can ensure singleton-workflows never execute more than one instance.
var lockKey = $"execute-workflow-definition:tenant:{tenantId}:workflow-definition:{workflowDefinitionId}";
await using var workflowDefinitionHandle = await AcquireLockAsync(lockKey, cancellationToken);
if (workflowBlueprint.IsSingleton)
{
_logger.LogDebug("Acquiring lock on correlation ID {CorrelationId}", correlationId);
correlationLockHandle = await _distributedLockProvider.AcquireLockAsync(correlationId, _elsaOptions.DistributedLockTimeout, cancellationToken);
if (correlationLockHandle == null)
throw new LockAcquisitionException($"Failed to acquire a lock on correlation ID {correlationId}");
}
try
{
// Acquire a lock on the workflow definition so that we can ensure singleton-workflows never execute more than one instance.
var lockKey = $"execute-workflow-definition:tenant:{tenantId}:workflow-definition:{workflowDefinitionId}";
await using var handle = await _distributedLockProvider.AcquireLockAsync(lockKey, _elsaOptions.DistributedLockTimeout, cancellationToken);
if (handle == null)
throw new LockAcquisitionException($"Failed to acquire a lock on {lockKey}");
if (workflowBlueprint.IsSingleton)
if (await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId))
{
if (await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId))
{
_logger.LogDebug("Workflow {WorkflowDefinitionId} is a singleton workflow and is already running");
return null;
}
}
var startActivities = _getsStartActivities.GetStartActivities(workflowBlueprint).Select(x => x.Id).ToHashSet();
var startActivityId = activityId == null ? startActivities.FirstOrDefault() : startActivities.Contains(activityId) ? activityId : default;
if (startActivityId == null)
{
_logger.LogWarning("Cannot start workflow {WorkflowDefinitionId} with version {WorkflowDefinitionVersion} because it has no starting activities", workflowBlueprint.Id, workflowBlueprint.Version);
_logger.LogDebug("Workflow {WorkflowDefinitionId} is a singleton workflow and is already running", workflowDefinitionId);
return null;
}
var workflowInstance = await _workflowFactory.InstantiateAsync(
workflowBlueprint,
correlationId,
contextId,
cancellationToken);
await _workflowInstanceStore.SaveAsync(workflowInstance, cancellationToken);
return new StartableWorkflow(workflowBlueprint, workflowInstance, startActivityId);
}
finally
var startActivities = _getsStartActivities.GetStartActivities(workflowBlueprint).Select(x => x.Id).ToHashSet();
var startActivityId = activityId == null ? startActivities.FirstOrDefault() : startActivities.Contains(activityId) ? activityId : default;
if (startActivityId == null)
{
if (correlationLockHandle != null)
await correlationLockHandle.DisposeAsync();
_logger.LogWarning("Cannot start workflow {WorkflowDefinitionId} with version {WorkflowDefinitionVersion} because it has no starting activities", workflowBlueprint.Id, workflowBlueprint.Version);
return null;
}
var workflowInstance = await _workflowFactory.InstantiateAsync(
workflowBlueprint,
correlationId,
contextId,
cancellationToken);
await _workflowInstanceStore.SaveAsync(workflowInstance, cancellationToken);
return new StartableWorkflow(workflowBlueprint, workflowInstance, startActivityId);
}
public async Task CollectAndExecuteStartableWorkflowAsync(string workflowDefinitionId, string? activityId, string? correlationId = default, string? contextId = default, object? input = default, string? tenantId = default, CancellationToken cancellationToken = default)
@ -263,30 +246,36 @@ namespace Elsa.Services
{
var correlationId = context.CorrelationId!;
var lockKey = correlationId;
await using var handle = await AcquireLockAsync(lockKey, cancellationToken);
await using (var handle = await _distributedLockProvider.AcquireLockAsync(lockKey, _elsaOptions.DistributedLockTimeout, cancellationToken))
var correlatedWorkflowInstanceCount = !string.IsNullOrWhiteSpace(correlationId)
? await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification<WorkflowInstance>(correlationId).WithStatus(WorkflowStatus.Suspended), cancellationToken)
: 0;
_logger.LogDebug("Found {CorrelatedWorkflowCount} workflows with correlation ID {CorrelationId}", correlatedWorkflowInstanceCount, correlationId);
if (correlatedWorkflowInstanceCount > 0)
{
if (handle == null)
throw new LockAcquisitionException($"Failed to acquire a lock on {lockKey}");
var correlatedWorkflowInstanceCount = !string.IsNullOrWhiteSpace(correlationId)
? await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification<WorkflowInstance>(correlationId).WithStatus(WorkflowStatus.Suspended), cancellationToken)
: 0;
_logger.LogDebug("Found {{CorrelatedWorkflowCount}} workflows with correlation ID {CorrelationId}", correlatedWorkflowInstanceCount, correlationId);
if (correlatedWorkflowInstanceCount > 0)
{
var bookmarkResults = context.Bookmark != null ? await _bookmarkFinder.FindBookmarksAsync(context.ActivityType, context.Bookmark, correlationId, context.TenantId, cancellationToken).ToList() : new List<BookmarkFinderResult>();
_logger.LogDebug("Found {BookmarkCount} bookmarks for activity type {ActivityType}", bookmarkResults.Count, context.ActivityType);
return bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList();
}
var bookmarkResults = context.Bookmark != null ? await _bookmarkFinder.FindBookmarksAsync(context.ActivityType, context.Bookmark, correlationId, context.TenantId, cancellationToken).ToList() : new List<BookmarkFinderResult>();
_logger.LogDebug("Found {BookmarkCount} bookmarks for activity type {ActivityType}", bookmarkResults.Count, context.ActivityType);
return bookmarkResults.Select(x => new PendingWorkflow(x.WorkflowInstanceId, x.ActivityId)).ToList();
}
var startableWorkflows = await CollectStartableWorkflowsAsync(context, cancellationToken);
return startableWorkflows.Select(x => new PendingWorkflow(x.WorkflowInstance.Id, x.ActivityId)).ToList();
}
private async Task<IDistributedSynchronizationHandle> AcquireLockAsync(string resource, CancellationToken cancellationToken)
{
var handle = await _distributedLockProvider.AcquireLockAsync(resource, _elsaOptions.DistributedLockTimeout, cancellationToken);
if (handle == null)
throw new LockAcquisitionException($"Failed to acquire a lock on {resource}");
return handle;
}
private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint)
{
if (workflowBlueprint != null)