From ea40ad1121bca58c778d4dc7654d150f528095cf Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 7 Jun 2021 15:21:36 +0200 Subject: [PATCH] More correlation fixes --- .../Bookmarks/QueueMessageReceivedBookmark.cs | 7 ++----- .../Services/QueueWorker.cs | 3 +-- .../Activities/EventReceived.cs | 4 ++-- .../Models/WorkflowInstance.cs | 3 +-- .../Services/IResumesWorkflows.cs | 2 -- .../Workflows/RunWorkflow/RunWorkflow.cs | 4 ++-- src/core/Elsa.Core/Services/WorkflowFactory.cs | 3 ++- .../Elsa.Core/Services/WorkflowLaunchpad.cs | 3 +-- src/core/Elsa.Core/Services/WorkflowResumer.cs | 4 ++-- .../StartupTasks/ContinueRunningWorkflows.cs | 18 +++++++++++++++--- 10 files changed, 28 insertions(+), 23 deletions(-) diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/QueueMessageReceivedBookmark.cs b/src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/QueueMessageReceivedBookmark.cs index 5d5104a55..8602df9d0 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/QueueMessageReceivedBookmark.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Bookmarks/QueueMessageReceivedBookmark.cs @@ -11,14 +11,12 @@ namespace Elsa.Activities.AzureServiceBus.Bookmarks { } - public QueueMessageReceivedBookmark(string queueName, string? correlationId = default) + public QueueMessageReceivedBookmark(string queueName) { QueueName = queueName; - CorrelationId = correlationId; } public string QueueName { get; set; } = default!; - public string? CorrelationId { get; set; } } public class QueueMessageReceivedBookmarkProvider : BookmarkProvider @@ -28,8 +26,7 @@ namespace Elsa.Activities.AzureServiceBus.Bookmarks { Result(new QueueMessageReceivedBookmark { - QueueName = (await context.ReadActivityPropertyAsync(x => x.QueueName, cancellationToken))!, - CorrelationId = context.ActivityExecutionContext.WorkflowExecutionContext.CorrelationId + QueueName = (await context.ReadActivityPropertyAsync(x => x.QueueName, cancellationToken))! }) }; } diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs index c0b704c26..938d57b4a 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs @@ -3,7 +3,6 @@ using System.Threading.Tasks; 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; @@ -25,7 +24,7 @@ namespace Elsa.Activities.AzureServiceBus.Services protected override string ActivityType => nameof(AzureServiceBusQueueMessageReceived); - protected override IBookmark CreateBookmark(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path, message.CorrelationId); + protected override IBookmark CreateBookmark(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path); protected override IBookmark CreateTrigger(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path); } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Conductor/Activities/EventReceived.cs b/src/activities/Elsa.Activities.Conductor/Activities/EventReceived.cs index cd74ad614..d76336c4c 100644 --- a/src/activities/Elsa.Activities.Conductor/Activities/EventReceived.cs +++ b/src/activities/Elsa.Activities.Conductor/Activities/EventReceived.cs @@ -30,7 +30,7 @@ namespace Elsa.Activities.Conductor DefaultSyntax = SyntaxNames.Json, SupportedSyntaxes = new[] { SyntaxNames.Json } )] - public ISet Outcomes { get; set; } = new HashSet(); + public ISet OutcomeNames { get; set; } = new HashSet(); [ActivityOutput(Hint = "Any input that was sent along with the event from your application.")] public object? Payload { get; set; } @@ -44,7 +44,7 @@ namespace Elsa.Activities.Conductor var outcomes = eventModel.Outcomes; if (outcomes?.Any() == false) - outcomes = new[] { OutcomeNames.Done }; + outcomes = new[] { Elsa.OutcomeNames.Done }; Payload = eventModel.Payload; return base.Outcomes(outcomes!); diff --git a/src/core/Elsa.Abstractions/Models/WorkflowInstance.cs b/src/core/Elsa.Abstractions/Models/WorkflowInstance.cs index 734fcdf65..7f5036242 100644 --- a/src/core/Elsa.Abstractions/Models/WorkflowInstance.cs +++ b/src/core/Elsa.Abstractions/Models/WorkflowInstance.cs @@ -1,4 +1,3 @@ -using System; using System.Collections.Generic; using Elsa.Comparers; using Newtonsoft.Json.Linq; @@ -21,7 +20,7 @@ namespace Elsa.Models public string? TenantId { get; set; } public int Version { get; set; } public WorkflowStatus WorkflowStatus { get; set; } - public string? CorrelationId { get; set; } + public string CorrelationId { get; set; } = default!; public string? ContextType { get; set; } public string? ContextId { get; set; } public string? Name { get; set; } diff --git a/src/core/Elsa.Abstractions/Services/IResumesWorkflows.cs b/src/core/Elsa.Abstractions/Services/IResumesWorkflows.cs index 5b43269bb..089adba32 100644 --- a/src/core/Elsa.Abstractions/Services/IResumesWorkflows.cs +++ b/src/core/Elsa.Abstractions/Services/IResumesWorkflows.cs @@ -10,8 +10,6 @@ namespace Elsa.Services Task ResumeWorkflowsAsync( IEnumerable results, object? input = default, - string? correlationId = default, - string? contextId = default, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Activities/Workflows/RunWorkflow/RunWorkflow.cs b/src/core/Elsa.Core/Activities/Workflows/RunWorkflow/RunWorkflow.cs index ca235106b..a40c31e02 100644 --- a/src/core/Elsa.Core/Activities/Workflows/RunWorkflow/RunWorkflow.cs +++ b/src/core/Elsa.Core/Activities/Workflows/RunWorkflow/RunWorkflow.cs @@ -90,8 +90,8 @@ namespace Elsa.Activities.Workflows return Outcome("Not Found"); var result = await _startsWorkflow.StartWorkflowAsync(workflowBlueprint!, TenantId, Input, CorrelationId, ContextId, cancellationToken); - var workflowInstance = result.WorkflowInstance; - var workflowStatus = result.WorkflowInstance.WorkflowStatus; + var workflowInstance = result.WorkflowInstance!; + var workflowStatus = result.WorkflowInstance!.WorkflowStatus; ChildWorkflowInstanceId = workflowInstance.Id; diff --git a/src/core/Elsa.Core/Services/WorkflowFactory.cs b/src/core/Elsa.Core/Services/WorkflowFactory.cs index b9048c2ca..29c010f05 100644 --- a/src/core/Elsa.Core/Services/WorkflowFactory.cs +++ b/src/core/Elsa.Core/Services/WorkflowFactory.cs @@ -1,3 +1,4 @@ +using System; using System.Threading; using System.Threading.Tasks; using Elsa.Models; @@ -30,7 +31,7 @@ namespace Elsa.Services TenantId = workflowBlueprint.TenantId, Version = workflowBlueprint.Version, WorkflowStatus = WorkflowStatus.Idle, - CorrelationId = correlationId, + CorrelationId = correlationId ?? Guid.NewGuid().ToString("N"), ContextId = contextId, CreatedAt = _clock.GetCurrentInstant(), Variables = new Variables(workflowBlueprint.Variables), diff --git a/src/core/Elsa.Core/Services/WorkflowLaunchpad.cs b/src/core/Elsa.Core/Services/WorkflowLaunchpad.cs index 39b35b94d..983bf12b7 100644 --- a/src/core/Elsa.Core/Services/WorkflowLaunchpad.cs +++ b/src/core/Elsa.Core/Services/WorkflowLaunchpad.cs @@ -274,9 +274,8 @@ namespace Elsa.Services private async Task> CollectResumableOrStartableCorrelatedWorkflowsAsync(CollectWorkflowsContext context, CancellationToken cancellationToken) { var correlationId = context.CorrelationId!; - var lockKey = correlationId; - await using var handle = await AcquireLockAsync(lockKey, cancellationToken); + await using var handle = await AcquireLockAsync(correlationId, cancellationToken); var correlatedWorkflowInstanceCount = !string.IsNullOrWhiteSpace(correlationId) ? await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification(correlationId).And(new WorkflowUnfinishedStatusSpecification()), cancellationToken) : 0; diff --git a/src/core/Elsa.Core/Services/WorkflowResumer.cs b/src/core/Elsa.Core/Services/WorkflowResumer.cs index c2d1d7fc2..9e6e14a12 100644 --- a/src/core/Elsa.Core/Services/WorkflowResumer.cs +++ b/src/core/Elsa.Core/Services/WorkflowResumer.cs @@ -39,10 +39,10 @@ namespace Elsa.Services CancellationToken cancellationToken = default) { var results = await _bookmarkFinder.FindBookmarksAsync(activityType, bookmark, correlationId, tenantId, cancellationToken).ToList(); - await ResumeWorkflowsAsync(results, input, correlationId, contextId, cancellationToken); + await ResumeWorkflowsAsync(results, input, cancellationToken); } - public async Task ResumeWorkflowsAsync(IEnumerable results, object? input = default, string? correlationId = default, string? contextId = default, CancellationToken cancellationToken = default) + public async Task ResumeWorkflowsAsync(IEnumerable results, object? input = default, CancellationToken cancellationToken = default) { foreach (var result in results) { diff --git a/src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs b/src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs index d1b372019..2190bbd02 100644 --- a/src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs +++ b/src/core/Elsa.Core/StartupTasks/ContinueRunningWorkflows.cs @@ -6,9 +6,11 @@ using Elsa.Models; using Elsa.Persistence; using Elsa.Persistence.Specifications.WorkflowInstances; using Elsa.Services; +using Medallion.Threading; using Microsoft.Extensions.Logging; using NodaTime; using Open.Linq.AsyncExtensions; +using IDistributedLockProvider = Elsa.Services.IDistributedLockProvider; namespace Elsa.StartupTasks { @@ -21,17 +23,20 @@ namespace Elsa.StartupTasks private readonly IWorkflowInstanceStore _workflowInstanceStore; private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher; private readonly IDistributedLockProvider _distributedLockProvider; + private readonly ElsaOptions _elsaOptions; private readonly ILogger _logger; public ContinueRunningWorkflows( IWorkflowInstanceStore workflowInstanceStore, IWorkflowInstanceDispatcher workflowInstanceDispatcher, IDistributedLockProvider distributedLockProvider, + ElsaOptions elsaOptions, ILogger logger) { _workflowInstanceStore = workflowInstanceStore; _workflowInstanceDispatcher = workflowInstanceDispatcher; _distributedLockProvider = distributedLockProvider; + _elsaOptions = elsaOptions; _logger = logger; } @@ -39,9 +44,7 @@ namespace Elsa.StartupTasks public async Task ExecuteAsync(CancellationToken cancellationToken = default) { - var lockKey = GetType().Name; - - await using var handle = await _distributedLockProvider.AcquireLockAsync(lockKey, Duration.FromSeconds(10), cancellationToken); + await using var handle = await _distributedLockProvider.AcquireLockAsync(GetType().Name, _elsaOptions.DistributedLockTimeout, cancellationToken); if (handle == null) return; @@ -55,6 +58,14 @@ namespace Elsa.StartupTasks foreach (var instance in instances) { + await using var correlationLockHandle = await _distributedLockProvider.AcquireLockAsync(instance.CorrelationId, _elsaOptions.DistributedLockTimeout, cancellationToken); + + if(handle == null) + { + _logger.LogWarning("Failed to acquire lock on correlation {CorrelationId} for workflow instance {WorkflowInstanceId}", instance.CorrelationId, instance.Id); + continue; + } + _logger.LogInformation("Resuming {WorkflowInstanceId}", instance.Id); var scheduledActivities = instance.ScheduledActivities; @@ -65,6 +76,7 @@ namespace Elsa.StartupTasks _logger.LogWarning( "Workflow '{WorkflowInstanceId}' was in the Running state, but has no scheduled activities not has a currently executing one. However, it does have blocking activities, so switching to Suspended status", instance.Id); + instance.WorkflowStatus = WorkflowStatus.Suspended; await _workflowInstanceStore.SaveAsync(instance, cancellationToken); continue;