From 6b68bd8f6db676d568eeb7beaf90259f068e8b4a Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sat, 12 Aug 2023 21:08:14 +0200 Subject: [PATCH] Prevent WorkflowInbox from triggering new workflows when targeting a specific workflow instance --- .../Services/ProtoActorWorkflowRuntime.cs | 2 +- .../Contracts/IWorkflowRuntime.cs | 2 +- .../Services/DefaultWorkflowInbox.cs | 50 +++++++++++-------- .../Services/DefaultWorkflowRuntime.cs | 7 +-- 4 files changed, 36 insertions(+), 25 deletions(-) diff --git a/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index 3b3d04058..31c0b7bda 100644 --- a/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -146,7 +146,7 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime } /// - public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default) + public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default) { var request = new ResumeWorkflowRequest { diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs index 9315f12bb..406d8d061 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs @@ -47,7 +47,7 @@ public interface IWorkflowRuntime /// The ID of the workflow instance to resume. /// /// - Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default); + Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default); /// /// Resumes all workflows that are bookmarked on the specified activity type. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs index 4ddc2890f..3b7af0ff2 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs @@ -25,11 +25,11 @@ public class DefaultWorkflowInbox : IWorkflowInbox /// Initializes a new instance of the class. /// public DefaultWorkflowInbox( - IWorkflowRuntime workflowRuntime, - IWorkflowInboxStore store, - INotificationSender notificationSender, - ISystemClock systemClock, - IIdentityGenerator identityGenerator, + IWorkflowRuntime workflowRuntime, + IWorkflowInboxStore store, + INotificationSender notificationSender, + ISystemClock systemClock, + IIdentityGenerator identityGenerator, IBookmarkHasher bookmarkHasher) { _workflowRuntime = workflowRuntime; @@ -51,7 +51,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox public async ValueTask SubmitAsync(NewWorkflowInboxMessage newMessage, WorkflowInboxMessageDeliveryOptions options, CancellationToken cancellationToken = default) { var now = _systemClock.UtcNow; - + // Create a new message. var message = new WorkflowInboxMessage { @@ -66,22 +66,38 @@ public class DefaultWorkflowInbox : IWorkflowInbox Input = newMessage.Input, Hash = _bookmarkHasher.Hash(newMessage.ActivityTypeName, newMessage.BookmarkPayload), }; - + // Store the message. await _store.SaveAsync(message, cancellationToken); - + // Send a notification. var strategy = options.EventPublishingStrategy; var workflowExecutionResults = new List(); var notification = new WorkflowInboxMessageReceived(message, workflowExecutionResults); await _notificationSender.SendAsync(notification, strategy, cancellationToken); - + // Return the result. return new SubmitWorkflowInboxMessageResult(message, workflowExecutionResults); } /// public async ValueTask DeliverAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) + { + var triggeredWorkflows = await TriggerWorkflowsAsync(message, cancellationToken); + + message.IsHandled = triggeredWorkflows.Any(); + + if (message.IsHandled) + { + message.AffectedWorkflowInstancesIds = triggeredWorkflows.Select(x => x.WorkflowInstanceId).ToList(); + message.HandledAt = _systemClock.UtcNow; + await _store.SaveAsync(message, cancellationToken); + } + + return new DeliverWorkflowInboxMessageResult(triggeredWorkflows); + } + + private async Task> TriggerWorkflowsAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) { var activityTypeName = message.ActivityTypeName; var correlationId = message.CorrelationId; @@ -90,18 +106,12 @@ public class DefaultWorkflowInbox : IWorkflowInbox var bookmarkPayload = message.BookmarkPayload; var input = message.Input; var options = new TriggerWorkflowsRuntimeOptions(correlationId, workflowInstanceId, activityInstanceId, input); - var result = await _workflowRuntime.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options, cancellationToken); - - message.IsHandled = result.TriggeredWorkflows.Any(); - if (message.IsHandled) - { - message.AffectedWorkflowInstancesIds = result.TriggeredWorkflows.Select(x => x.WorkflowInstanceId).ToList(); - message.HandledAt = _systemClock.UtcNow; - await _store.SaveAsync(message, cancellationToken); - } - - return new DeliverWorkflowInboxMessageResult(result.TriggeredWorkflows); + if (workflowInstanceId != null) + return await _workflowRuntime.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options, cancellationToken); + + var result = await _workflowRuntime.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options, cancellationToken); + return result.TriggeredWorkflows; } /// diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs index 6b6bcb8ab..d857d149f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs @@ -119,14 +119,14 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime } /// - public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default) + public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default) { await using (await _distributedLockProvider.AcquireLockAsync(workflowInstanceId, TimeSpan.FromMinutes(1), cancellationToken)) { var workflowState = await _workflowStateStore.FindAsync(workflowInstanceId, cancellationToken); if (workflowState == null) - throw new Exception($"Workflow instance {workflowInstanceId} not found"); + return null; var definitionId = workflowState.DefinitionId; var version = workflowState.DefinitionVersion; @@ -309,7 +309,8 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime }, cancellationToken); - resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Status, resumeResult.SubStatus, resumeResult.Bookmarks)); + if (resumeResult != null) + resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Status, resumeResult.SubStatus, resumeResult.Bookmarks)); } return resumedWorkflows;