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

Fix locking bug

Browse source
This commit is contained in:
Sipke Schoorstra 2021-04-09 22:56:07 +02:00
parent 08bc052a03
commit 7991d4a468
1 changed files with 14 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
src/core/Elsa.Core/Dispatch/Handlers/ExecuteWorkflowDefinition.cs
View file

@ -57,38 +57,39 @@ namespace Elsa.Dispatch.Handlers
{
_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}");
}
if(correlationLockHandle == null)
throw new LockAcquisitionException($"Failed to acquire a lock on correlation ID {correlationId}");
try
{
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 || await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId) == false)
await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken);
if (handle == null)
throw new LockAcquisitionException($"Failed to acquire a lock on {lockKey}");
if (!workflowBlueprint!.IsSingleton || await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId) == false)
await _startsWorkflow.StartWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken);
}
finally
{
await correlationLockHandle.DisposeAsync();
if (correlationLockHandle != null)
await correlationLockHandle.DisposeAsync();
}
return Unit.Value;
}
private bool ValidatePreconditions(string? workflowDefinitionId, IWorkflowBlueprint? workflowBlueprint)
{
if (workflowBlueprint != null)
if (workflowBlueprint != null)
return true;
_logger.LogWarning("No workflow definition {WorkflowDefinitionId} found. Make sure the scheduled workflow definition is published and enabled", workflowDefinitionId);
return false;
}
private async Task<bool> GetWorkflowIsAlreadyExecutingAsync(string? tenantId, string workflowDefinitionId)
{
var specification = new TenantSpecification<WorkflowInstance>(tenantId).WithWorkflowDefinition(workflowDefinitionId).And(new WorkflowIsAlreadyExecutingSpecification());