From fbd16f2d747a9e6c239d12fd32d3bdcfb5c92d4c Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Fri, 24 Jan 2025 13:54:37 +0100 Subject: [PATCH 1/4] Refactor workflow execution handling to improve performance --- .../Activities/BulkDispatchWorkflows.cs | 49 ++++++++++++------- .../Activities/DispatchWorkflow.cs | 35 ++++++++----- .../Activities/ExecuteWorkflow.cs | 40 +++++++++------ .../ResumeBulkDispatchWorkflowActivity.cs | 9 ++-- .../ResumeDispatchWorkflowActivity.cs | 18 +++++-- .../Handlers/ResumeExecuteWorkflowActivity.cs | 14 ++++-- 6 files changed, 109 insertions(+), 56 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index df335a24d..ca1ccd2ae 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -104,19 +104,11 @@ public class BulkDispatchWorkflows : Activity protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { var waitForCompletion = WaitForCompletion.GetOrDefault(context); - var items = context.GetItemSource(Items); - var dispatchedInstancesCount = 0; - - await foreach (var item in items) - { - await DispatchChildWorkflowAsync(context, item); - dispatchedInstancesCount++; - } - - context.SetProperty(DispatchedInstancesCountKey, dispatchedInstancesCount); + var items = await context.GetItemSource(Items).ToListAsync(context.CancellationToken); + var count = items.Count; // If we need to wait for the child workflows to complete (if any), create a bookmark. - if (waitForCompletion && dispatchedInstancesCount > 0) + if (waitForCompletion && count > 0) { var workflowInstanceId = context.WorkflowExecutionContext.Id; var bookmarkOptions = new CreateBookmarkArgs @@ -125,23 +117,45 @@ public class BulkDispatchWorkflows : Activity Stimulus = new BulkDispatchWorkflowsStimulus(workflowInstanceId) { ParentInstanceId = context.WorkflowExecutionContext.Id, - ScheduledInstanceIdsCount = dispatchedInstancesCount + ScheduledInstanceIdsCount = count }, IncludeActivityInstanceId = false, AutoBurn = false, }; + + // Create bookmarks first. context.CreateBookmark(bookmarkOptions); + + // Dispatch workflows afterwards. + await DispatchWorkflowsAsync(); } else { // Otherwise, we can complete immediately. + await DispatchWorkflowsAsync(); await context.CompleteActivityWithOutcomesAsync("Done"); } + + return; + + async Task DispatchWorkflowsAsync() + { + foreach (var item in items) + await DispatchChildWorkflowAsync(context, item, waitForCompletion); + + context.SetProperty(DispatchedInstancesCountKey, count); + } } - private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context, object item) + private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context, object item, bool waitForCompletion) { var workflowDefinitionId = WorkflowDefinitionId.Get(context); + var workflowDefinitionService = context.GetRequiredService(); + var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published); + + if (workflowGraph == null) + throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); + var parentInstanceId = context.WorkflowExecutionContext.Id; var input = Input.GetOrDefault(context) ?? new Dictionary(); var channelName = ChannelName.GetOrDefault(context); @@ -150,6 +164,9 @@ public class BulkDispatchWorkflows : Activity { ["ParentInstanceId"] = parentInstanceId }; + + if(waitForCompletion) + properties["WaitForCompletion"] = true; var itemDictionary = new Dictionary { @@ -168,12 +185,6 @@ public class BulkDispatchWorkflows : Activity var workflowDispatcher = context.GetRequiredService(); var identityGenerator = context.GetRequiredService(); var evaluator = context.GetRequiredService(); - var workflowDefinitionService = context.GetRequiredService(); - var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published); - - if (workflowGraph == null) - throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found."); - var correlationId = CorrelationIdFunction != null ? await evaluator.EvaluateAsync(CorrelationIdFunction!, context.ExpressionExecutionContext, evaluatorOptions) : null; var instanceId = identityGenerator.GenerateId(); var request = new DispatchWorkflowDefinitionRequest(workflowGraph.Workflow.Identity.Id) diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs index fa424c91d..4e579d60c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs @@ -72,7 +72,7 @@ public class DispatchWorkflow : Activity var waitForCompletion = WaitForCompletion.GetOrDefault(context); // Dispatch the child workflow. - var instanceId = await DispatchChildWorkflowAsync(context); + var instanceId = await DispatchChildWorkflowAsync(context, waitForCompletion); // If we need to wait for the child workflow to complete, create a bookmark. if (waitForCompletion) @@ -92,28 +92,39 @@ public class DispatchWorkflow : Activity } } - private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context) + private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context, bool waitForCompletion) { var workflowDefinitionId = WorkflowDefinitionId.Get(context); - var input = Input.GetOrDefault(context) ?? new Dictionary(); - var channelName = ChannelName.GetOrDefault(context); - - input["ParentInstanceId"] = context.WorkflowExecutionContext.Id; - - var correlationId = CorrelationId.GetOrDefault(context); - var workflowDispatcher = context.GetRequiredService(); - var identityGenerator = context.GetRequiredService(); var workflowDefinitionService = context.GetRequiredService(); var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken); if (workflowGraph == null) - throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found."); + throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); + + var input = Input.GetOrDefault(context) ?? new Dictionary(); + var channelName = ChannelName.GetOrDefault(context); + var parentInstanceId = context.WorkflowExecutionContext.Id; + var properties = new Dictionary + { + ["ParentInstanceId"] = parentInstanceId + }; + + if(waitForCompletion) + properties["WaitForCompletion"] = true; + + input["ParentInstanceId"] = parentInstanceId; + + var correlationId = CorrelationId.GetOrDefault(context); + var workflowDispatcher = context.GetRequiredService(); + var identityGenerator = context.GetRequiredService(); + var instanceId = identityGenerator.GenerateId(); var request = new DispatchWorkflowDefinitionRequest(workflowGraph.Workflow.Identity.Id) { - ParentWorkflowInstanceId = context.WorkflowExecutionContext.Id, + ParentWorkflowInstanceId = parentInstanceId, Input = input, + Properties = properties, CorrelationId = correlationId, InstanceId = instanceId, }; diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs index 7aaf05a5b..cba602166 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs @@ -19,7 +19,7 @@ namespace Elsa.Workflows.Runtime.Activities; public class ExecuteWorkflow : Activity { /// - public ExecuteWorkflow([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + public ExecuteWorkflow([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line) { } @@ -31,7 +31,7 @@ public class ExecuteWorkflow : Activity Description = "The definition ID of the workflow to execute.", UIHint = InputUIHints.WorkflowDefinitionPicker )] - public Input WorkflowDefinitionId { get; set; } = default!; + public Input WorkflowDefinitionId { get; set; } = null!; /// /// The correlation ID to associate the workflow with. @@ -40,25 +40,25 @@ public class ExecuteWorkflow : Activity DisplayName = "Correlation ID", Description = "The correlation ID to associate the workflow with." )] - public Input CorrelationId { get; set; } = default!; + public Input CorrelationId { get; set; } = null!; /// /// The input to send to the workflow. /// [Input(Description = "The input to send to the workflow.")] - public Input?> Input { get; set; } = default!; + public Input?> Input { get; set; } = null!; /// /// True to wait for the child workflow to complete before completing this activity. If not set, the child workflow will be executed until it either completes or goes idle before this activity completes. /// [Input(Description = "Wait for the child workflow to complete before completing this activity.")] - public Input WaitForCompletion { get; set; } = default!; + public Input WaitForCompletion { get; set; } = null!; /// protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { - var result = await ExecuteWorkflowAsync(context); var waitForCompletion = WaitForCompletion.Get(context); + var result = await ExecuteWorkflowAsync(context, waitForCompletion); if(!waitForCompletion || result.Status == WorkflowStatus.Finished) { @@ -77,23 +77,35 @@ public class ExecuteWorkflow : Activity context.CreateBookmark(bookmarkOptions); } - private async ValueTask ExecuteWorkflowAsync(ActivityExecutionContext context) + private async ValueTask ExecuteWorkflowAsync(ActivityExecutionContext context, bool waitForCompletion) { var workflowDefinitionId = WorkflowDefinitionId.Get(context); - var input = Input.GetOrDefault(context) ?? new Dictionary(); - var correlationId = CorrelationId.GetOrDefault(context); - var workflowInvoker = context.GetRequiredService(); - var identityGenerator = context.GetRequiredService(); var workflowDefinitionService = context.GetRequiredService(); var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken); if (workflowGraph == null) - throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found."); - + throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); + + var parentInstanceId = context.WorkflowExecutionContext.Id; + var input = Input.GetOrDefault(context) ?? new Dictionary(); + var correlationId = CorrelationId.GetOrDefault(context); + var workflowInvoker = context.GetRequiredService(); + var identityGenerator = context.GetRequiredService(); + var properties = new Dictionary + { + ["ParentInstanceId"] = parentInstanceId + }; + + if(waitForCompletion) + properties["WaitForCompletion"] = true; + + input["ParentInstanceId"] = parentInstanceId; + var options = new RunWorkflowOptions { - ParentWorkflowInstanceId = context.WorkflowExecutionContext.Id, + ParentWorkflowInstanceId = parentInstanceId, Input = input, + Properties = properties, CorrelationId = correlationId, WorkflowInstanceId = identityGenerator.GenerateId() }; diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs index 6b683305b..7532d376e 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs @@ -21,11 +21,12 @@ internal class ResumeBulkDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue, if (workflowState.Status != WorkflowStatus.Finished) return; - var parentInstanceId = workflowState.Properties.TryGetValue("ParentInstanceId", out var parentInstanceIdValue) ? parentInstanceIdValue.ToString() : default; - - if (string.IsNullOrWhiteSpace(parentInstanceId)) + var waitForCompletion = workflowState.Properties.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue; + + if (!waitForCompletion) return; - + + var parentInstanceId = (string)workflowState.Properties["ParentInstanceId"]; var activityTypeName = ActivityTypeNameHelper.GenerateTypeName(); var stimulus = new BulkDispatchWorkflowsStimulus(parentInstanceId); var stimulusHash = stimulusHasher.Hash(activityTypeName, stimulus); diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs index 607edf1f6..f1b52b181 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs @@ -2,7 +2,6 @@ using Elsa.Mediator.Contracts; using Elsa.Workflows.Helpers; using Elsa.Workflows.Notifications; using Elsa.Workflows.Runtime.Activities; -using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Stimuli; using JetBrains.Annotations; using Microsoft.Extensions.Logging; @@ -22,21 +21,32 @@ internal class ResumeDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue, ISti var workflowState = notification.WorkflowState; logger.LogDebug("Handling workflow executed notification for workflow {WorkflowInstanceId}", notification.WorkflowState.Id); - + if (workflowState.Status != WorkflowStatus.Finished) { logger.LogDebug("Workflow {WorkflowInstanceId} is not in a finished state. Skipping resumption of any blocking DispatchWorkflow activities", notification.WorkflowState.Id); return; } + var props = workflowState.Properties; + var waitForCompletion = props.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue; + + if (!waitForCompletion) + { + logger.LogDebug("Workflow {WorkflowInstanceId} does not have a WaitForCompletion property set to true. Skipping resumption of any blocking DispatchWorkflow activities", notification.WorkflowState.Id); + return; + } + + var parentInstanceId = (string) props["ParentInstanceId"]; var stimulus = new DispatchWorkflowStimulus(notification.WorkflowState.Id); var input = workflowState.Output; - + var bookmarkQueueItem = new NewBookmarkQueueItem { + WorkflowInstanceId = parentInstanceId, ActivityTypeName = ActivityTypeName, StimulusHash = stimulusHasher.Hash(ActivityTypeName, stimulus), - Options = new ResumeBookmarkOptions + Options = new() { Input = input } diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs index 256621e0b..d4bb203a6 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs @@ -2,7 +2,6 @@ using Elsa.Mediator.Contracts; using Elsa.Workflows.Helpers; using Elsa.Workflows.Notifications; using Elsa.Workflows.Runtime.Activities; -using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Stimuli; using JetBrains.Annotations; @@ -11,7 +10,7 @@ namespace Elsa.Workflows.Runtime.Handlers; /// /// Resumes any blocking activities when its child workflow completes. /// -[PublicAPI] +[UsedImplicitly] internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStimulusHasher stimulusHasher) : INotificationHandler { private static readonly string ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName(); @@ -23,14 +22,23 @@ internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStim if (workflowState.Status != WorkflowStatus.Finished) return; + var props = workflowState.Properties; + + var waitForCompletion = props.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue; + + if (!waitForCompletion) + return; + + var parentInstanceId = (string)props["ParentInstanceId"]; var stimulus = new ExecuteWorkflowStimulus(notification.WorkflowState.Id); var input = workflowState.Output; var bookmarkQueueItem = new NewBookmarkQueueItem { + WorkflowInstanceId = parentInstanceId, ActivityTypeName = ActivityTypeName, StimulusHash = stimulusHasher.Hash(ActivityTypeName, stimulus), - Options = new ResumeBookmarkOptions + Options = new() { Input = input } From 74a5c3cd87c6551464e1f9bc440d925358aa5398 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Fri, 24 Jan 2025 13:58:50 +0100 Subject: [PATCH 2/4] Refactor workflow dispatching for clarity and efficiency Refactored the handling of child workflow execution, ensuring better readability and streamlining logic for dispatch operations. Added explicit tracking of dispatched instances and improved comments to support maintainability. --- .../Activities/BulkDispatchWorkflows.cs | 30 +++++++------------ .../Activities/DispatchWorkflow.cs | 11 ++++--- .../Activities/ExecuteWorkflow.cs | 21 ++++++------- 3 files changed, 27 insertions(+), 35 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index ca1ccd2ae..8074be822 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -107,6 +107,13 @@ public class BulkDispatchWorkflows : Activity var items = await context.GetItemSource(Items).ToListAsync(context.CancellationToken); var count = items.Count; + // Dispatch the child workflows. + foreach (var item in items) + await DispatchChildWorkflowAsync(context, item, waitForCompletion); + + // Store the number of dispatched instances for tracking. + context.SetProperty(DispatchedInstancesCountKey, count); + // If we need to wait for the child workflows to complete (if any), create a bookmark. if (waitForCompletion && count > 0) { @@ -122,29 +129,14 @@ public class BulkDispatchWorkflows : Activity IncludeActivityInstanceId = false, AutoBurn = false, }; - - // Create bookmarks first. + context.CreateBookmark(bookmarkOptions); - - // Dispatch workflows afterwards. - await DispatchWorkflowsAsync(); } else { // Otherwise, we can complete immediately. - await DispatchWorkflowsAsync(); await context.CompleteActivityWithOutcomesAsync("Done"); } - - return; - - async Task DispatchWorkflowsAsync() - { - foreach (var item in items) - await DispatchChildWorkflowAsync(context, item, waitForCompletion); - - context.SetProperty(DispatchedInstancesCountKey, count); - } } private async ValueTask DispatchChildWorkflowAsync(ActivityExecutionContext context, object item, bool waitForCompletion) @@ -155,7 +147,7 @@ public class BulkDispatchWorkflows : Activity if (workflowGraph == null) throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); - + var parentInstanceId = context.WorkflowExecutionContext.Id; var input = Input.GetOrDefault(context) ?? new Dictionary(); var channelName = ChannelName.GetOrDefault(context); @@ -164,8 +156,8 @@ public class BulkDispatchWorkflows : Activity { ["ParentInstanceId"] = parentInstanceId }; - - if(waitForCompletion) + + if (waitForCompletion) properties["WaitForCompletion"] = true; var itemDictionary = new Dictionary diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs index 4e579d60c..566c79937 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs @@ -97,10 +97,10 @@ public class DispatchWorkflow : Activity var workflowDefinitionId = WorkflowDefinitionId.Get(context); var workflowDefinitionService = context.GetRequiredService(); var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken); - + if (workflowGraph == null) throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); - + var input = Input.GetOrDefault(context) ?? new Dictionary(); var channelName = ChannelName.GetOrDefault(context); var parentInstanceId = context.WorkflowExecutionContext.Id; @@ -108,8 +108,9 @@ public class DispatchWorkflow : Activity { ["ParentInstanceId"] = parentInstanceId }; - - if(waitForCompletion) + + // If we need to wait for the child workflow to complete, set the property. This will be used by the ResumeDispatchWorkflowActivity handler. + if (waitForCompletion) properties["WaitForCompletion"] = true; input["ParentInstanceId"] = parentInstanceId; @@ -117,8 +118,6 @@ public class DispatchWorkflow : Activity var correlationId = CorrelationId.GetOrDefault(context); var workflowDispatcher = context.GetRequiredService(); var identityGenerator = context.GetRequiredService(); - - var instanceId = identityGenerator.GenerateId(); var request = new DispatchWorkflowDefinitionRequest(workflowGraph.Workflow.Identity.Id) { diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs index cba602166..4fda5e880 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs @@ -47,7 +47,7 @@ public class ExecuteWorkflow : Activity /// [Input(Description = "The input to send to the workflow.")] public Input?> Input { get; set; } = null!; - + /// /// True to wait for the child workflow to complete before completing this activity. If not set, the child workflow will be executed until it either completes or goes idle before this activity completes. /// @@ -59,14 +59,14 @@ public class ExecuteWorkflow : Activity { var waitForCompletion = WaitForCompletion.Get(context); var result = await ExecuteWorkflowAsync(context, waitForCompletion); - - if(!waitForCompletion || result.Status == WorkflowStatus.Finished) + + if (!waitForCompletion || result.Status == WorkflowStatus.Finished) { context.SetResult(result); await context.CompleteActivityAsync(); return; } - + // Since the child workflow is still running, we need to wait for it to complete using a bookmark. var bookmarkOptions = new CreateBookmarkArgs { @@ -85,7 +85,7 @@ public class ExecuteWorkflow : Activity if (workflowGraph == null) throw new($"No published version of workflow definition with ID {workflowDefinitionId} found."); - + var parentInstanceId = context.WorkflowExecutionContext.Id; var input = Input.GetOrDefault(context) ?? new Dictionary(); var correlationId = CorrelationId.GetOrDefault(context); @@ -95,12 +95,13 @@ public class ExecuteWorkflow : Activity { ["ParentInstanceId"] = parentInstanceId }; - - if(waitForCompletion) + + // If we need to wait for the child workflow to complete, set the property. This will be used by the ResumeExecuteWorkflowActivity to resume the parent workflow. + if (waitForCompletion) properties["WaitForCompletion"] = true; - + input["ParentInstanceId"] = parentInstanceId; - + var options = new RunWorkflowOptions { ParentWorkflowInstanceId = parentInstanceId, @@ -121,7 +122,7 @@ public class ExecuteWorkflow : Activity return info; } - + private async ValueTask OnChildWorkflowCompletedAsync(ActivityExecutionContext context) { var input = context.WorkflowInput; From 95e04eac1726889298b55349f8da01d61cfe3c14 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Fri, 24 Jan 2025 14:01:19 +0100 Subject: [PATCH 3/4] Refactor to remove redundant debug logs and clean up code. Eliminated unnecessary debug log statements from `ResumeDispatchWorkflowActivity` and removed redundant whitespace across the affected files. These changes streamline the code for better readability and maintainability --- .../ResumeDispatchWorkflowActivity.cs | 23 ++++++------------- .../Handlers/ResumeExecuteWorkflowActivity.cs | 3 +-- 2 files changed, 8 insertions(+), 18 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs index f1b52b181..37b95583a 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs @@ -15,32 +15,24 @@ namespace Elsa.Workflows.Runtime.Handlers; internal class ResumeDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue, IStimulusHasher stimulusHasher, ILogger logger) : INotificationHandler { private static readonly string ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName(); - + public async Task HandleAsync(WorkflowExecuted notification, CancellationToken cancellationToken) { var workflowState = notification.WorkflowState; - - logger.LogDebug("Handling workflow executed notification for workflow {WorkflowInstanceId}", notification.WorkflowState.Id); - + if (workflowState.Status != WorkflowStatus.Finished) - { - logger.LogDebug("Workflow {WorkflowInstanceId} is not in a finished state. Skipping resumption of any blocking DispatchWorkflow activities", notification.WorkflowState.Id); return; - } var props = workflowState.Properties; var waitForCompletion = props.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue; - + if (!waitForCompletion) - { - logger.LogDebug("Workflow {WorkflowInstanceId} does not have a WaitForCompletion property set to true. Skipping resumption of any blocking DispatchWorkflow activities", notification.WorkflowState.Id); return; - } - - var parentInstanceId = (string) props["ParentInstanceId"]; + + var parentInstanceId = (string)props["ParentInstanceId"]; var stimulus = new DispatchWorkflowStimulus(notification.WorkflowState.Id); var input = workflowState.Output; - + var bookmarkQueueItem = new NewBookmarkQueueItem { WorkflowInstanceId = parentInstanceId, @@ -51,8 +43,7 @@ internal class ResumeDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue, ISti Input = input } }; - - logger.LogDebug("Resuming any blocking DispatchWorkflow activities for workflow {WorkflowInstanceId} using stimulus hash {StimulusHash}", notification.WorkflowState.Id, bookmarkQueueItem.StimulusHash); + await bookmarkQueue.EnqueueAsync(bookmarkQueueItem, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs index d4bb203a6..da9ffefd9 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs @@ -23,9 +23,8 @@ internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStim return; var props = workflowState.Properties; - var waitForCompletion = props.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue; - + if (!waitForCompletion) return; From 184b159c6fd09d7fda5dd350531076be0127ec5e Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Fri, 24 Jan 2025 14:22:16 +0100 Subject: [PATCH 4/4] Bump base version to 3.3.1 in workflow configuration --- .github/workflows/packages.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/packages.yml b/.github/workflows/packages.yml index df4fbec7c..05ae4fb20 100644 --- a/.github/workflows/packages.yml +++ b/.github/workflows/packages.yml @@ -9,7 +9,7 @@ on: release: types: [ prereleased, published ] env: - base_version: '3.3.0' + base_version: '3.3.1' feedz_feed_source: 'https://f.feedz.io/elsa-workflows/elsa-3/nuget/index.json' nuget_feed_source: 'https://api.nuget.org/v3/index.json'