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

Apply distributed lock when triggering new workflows

Browse source
This commit is contained in:
Sipke Schoorstra 2022-12-21 20:25:25 +01:00
parent 193af7d9c8
commit 3755ff3f4d
1 changed files with 18 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
src/modules/Elsa.Workflows.Runtime/Implementations/DefaultWorkflowRuntime.cs
View file

@ -122,21 +122,25 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
var triggeredWorkflows = new List<TriggeredWorkflow>();
var hash = _hasher.Hash(activityTypeName, bookmarkPayload);
// Start new workflows.
var triggers = await _triggerStore.FindAsync(hash, cancellationToken);
foreach (var trigger in triggers)
// Start new workflows. Notice that this happens in a process-synchronized fashion to avoid multiple instances being created.
const string sharedResource = $"{nameof(DefaultWorkflowRuntime)}__StartTriggeredWorkflows";
await using (await _distributedLockProvider.AcquireLockAsync(sharedResource, TimeSpan.FromMinutes(10), cancellationToken))
{
var definitionId = trigger.WorkflowDefinitionId;
var startOptions = new StartWorkflowRuntimeOptions(options.CorrelationId, options.Input, VersionOptions.Published, trigger.ActivityId);
var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions, cancellationToken);
// If we can't start the workflow, don't try it.
if(!canStartResult.CanStart)
continue;
var startResult = await StartWorkflowAsync(definitionId, startOptions, cancellationToken);
triggeredWorkflows.Add(new TriggeredWorkflow(startResult.InstanceId, startResult.Bookmarks));
var triggers = await _triggerStore.FindAsync(hash, cancellationToken);
foreach (var trigger in triggers)
{
var definitionId = trigger.WorkflowDefinitionId;
var startOptions = new StartWorkflowRuntimeOptions(options.CorrelationId, options.Input, VersionOptions.Published, trigger.ActivityId);
var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions, cancellationToken);
// If we can't start the workflow, don't try it.
if (!canStartResult.CanStart)
continue;
var startResult = await StartWorkflowAsync(definitionId, startOptions, cancellationToken);
triggeredWorkflows.Add(new TriggeredWorkflow(startResult.InstanceId, startResult.Bookmarks));
}
}
// Resume bookmarks.