From d74159d59466f774be8bae7a128e7254d7921ad4 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 26 Nov 2024 20:35:41 +0100 Subject: [PATCH 1/3] Add bookmark queue purging and ExecuteWorkflow stimulus Introduce BookmarkQueuePurgeOptions and ExecuteWorkflowStimulus classes for managing workflow queue purging and stimulus handling. Implement related handler and activity modifications to support waiting for child workflows and purging old bookmark queue entries. --- src/apps/Elsa.Server.Web/Program.cs | 5 ++- .../Activities/ExecuteWorkflow.cs | 33 ++++++++++++++- .../Features/WorkflowRuntimeFeature.cs | 11 ++++- .../Handlers/ResumeExecuteWorkflowActivity.cs | 42 +++++++++++++++++++ .../Options/BookmarkQueuePurgeOptions.cs | 17 ++++++++ .../Services/DefaultBookmarkQueuePurger.cs | 11 +++-- .../Stimuli/ExecuteWorkflowStimulus.cs | 9 ++++ 7 files changed, 117 insertions(+), 11 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Stimuli/ExecuteWorkflowStimulus.cs diff --git a/src/apps/Elsa.Server.Web/Program.cs b/src/apps/Elsa.Server.Web/Program.cs index 14e416f61..ecb451afd 100644 --- a/src/apps/Elsa.Server.Web/Program.cs +++ b/src/apps/Elsa.Server.Web/Program.cs @@ -38,6 +38,7 @@ using Elsa.Workflows.LogPersistence; using Elsa.Workflows.Management.Compression; using Elsa.Workflows.Management.Stores; using Elsa.Workflows.Runtime.Distributed.Extensions; +using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Stores; using Elsa.Workflows.Runtime.Tasks; using JetBrains.Annotations; @@ -527,9 +528,11 @@ services.Configure(options => { options.Schedule.ConfigureTask(TimeSpan.FromSeconds(30)); options.Schedule.ConfigureTask(TimeSpan.FromHours(4)); - options.Schedule.ConfigureTask(TimeSpan.FromSeconds(60)); + options.Schedule.ConfigureTask(TimeSpan.FromSeconds(11)); }); +services.Configure(options => options.Ttl = TimeSpan.FromSeconds(10)); + //services.Configure(options => options.CacheDuration = TimeSpan.FromDays(1)); services.AddHealthChecks(); services.AddControllers(); diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs index bc23ea78a..7aaf05a5b 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs @@ -5,6 +5,7 @@ using Elsa.Workflows.Attributes; using Elsa.Workflows.Management; using Elsa.Workflows.Models; using Elsa.Workflows.Options; +using Elsa.Workflows.Runtime.Stimuli; using Elsa.Workflows.UIHints; using JetBrains.Annotations; @@ -46,13 +47,34 @@ public class ExecuteWorkflow : Activity /// [Input(Description = "The input to send to the workflow.")] public Input?> Input { get; set; } = default!; + + /// + /// 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!; /// protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { var result = await ExecuteWorkflowAsync(context); - context.SetResult(result); - await context.CompleteActivityAsync(); + var waitForCompletion = WaitForCompletion.Get(context); + + 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 + { + Callback = OnChildWorkflowCompletedAsync, + Stimulus = new ExecuteWorkflowStimulus(result.WorkflowInstanceId), + IncludeActivityInstanceId = false + }; + context.CreateBookmark(bookmarkOptions); } private async ValueTask ExecuteWorkflowAsync(ActivityExecutionContext context) @@ -87,4 +109,11 @@ public class ExecuteWorkflow : Activity return info; } + + private async ValueTask OnChildWorkflowCompletedAsync(ActivityExecutionContext context) + { + var input = context.WorkflowInput; + context.Set(Result, input); + await context.CompleteActivityAsync(); + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index 5adcb1735..7f9d301d4 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -137,6 +137,11 @@ public class WorkflowRuntimeFeature : FeatureBase /// A delegate to configure the . /// public Action WorkflowDispatcherOptions { get; set; } = _ => { }; + + /// + /// A delegate to configure the . + /// + public Action BookmarkQueuePurgeOptions { get; set; } = _ => { }; /// /// Register the specified workflow type. @@ -205,6 +210,7 @@ public class WorkflowRuntimeFeature : FeatureBase Services.Configure(DistributedLockingOptions); Services.Configure(WorkflowInboxCleanupOptions); Services.Configure(WorkflowDispatcherOptions); + Services.Configure(BookmarkQueuePurgeOptions); Services.Configure(options => { options.Workflows = Workflows; }); Services.Configure(options => { @@ -292,7 +298,7 @@ public class WorkflowRuntimeFeature : FeatureBase // Startup tasks, background tasks, and recurring tasks. .AddStartupTask() .AddRecurringTask(TimeSpan.FromMinutes(1)) - .AddRecurringTask(TimeSpan.FromMinutes(1)) + .AddRecurringTask(TimeSpan.FromSeconds(10)) // Distributed locking. .AddSingleton(DistributedLockProvider) @@ -300,13 +306,14 @@ public class WorkflowRuntimeFeature : FeatureBase // Workflow definition providers. .AddWorkflowDefinitionProvider() - // UI prooprty handlers. + // UI property handlers. .AddScoped() // Domain handlers. .AddCommandHandler() .AddNotificationHandler() .AddNotificationHandler() + .AddNotificationHandler() .AddNotificationHandler() .AddNotificationHandler() .AddNotificationHandler() diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs new file mode 100644 index 000000000..9d4912d38 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs @@ -0,0 +1,42 @@ +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; + +namespace Elsa.Workflows.Runtime.Handlers; + +/// +/// Resumes any blocking activities when its child workflow completes. +/// +[PublicAPI] +internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStimulusHasher stimulusHasher) : INotificationHandler +{ + private static readonly string ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName(); + + public async Task HandleAsync(WorkflowExecuted notification, CancellationToken cancellationToken) + { + var workflowState = notification.WorkflowState; + + if (workflowState.Status != WorkflowStatus.Finished) + return; + + var stimulus = new ExecuteWorkflowStimulus(notification.WorkflowState.Id); + var input = workflowState.Output; + + var bookmarkQueueItem = new NewBookmarkQueueItem + { + ActivityTypeName = ActivityTypeName, + StimulusHash = stimulusHasher.Hash(ActivityTypeName, stimulus), + Options = new ResumeBookmarkOptions + { + Input = input + } + }; + + await bookmarkQueue.EnqueueAsync(bookmarkQueueItem, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs new file mode 100644 index 000000000..0c765563a --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs @@ -0,0 +1,17 @@ +namespace Elsa.Workflows.Runtime.Options; + +/// +/// Options for purging the bookmark queue. +/// +public class BookmarkQueuePurgeOptions +{ + /// + /// The time-to-live for bookmark queue items. + /// + public TimeSpan Ttl { get; set; } = TimeSpan.FromSeconds(10); + + /// + /// The number of records to clean up per sweep. + /// + public int BatchSize { get; set; } = 1000; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBookmarkQueuePurger.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBookmarkQueuePurger.cs index a1ce6110c..82a62e055 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBookmarkQueuePurger.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBookmarkQueuePurger.cs @@ -2,29 +2,28 @@ using Elsa.Common; using Elsa.Common.Entities; using Elsa.Common.Models; using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.OrderDefinitions; using JetBrains.Annotations; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; namespace Elsa.Workflows.Runtime; [UsedImplicitly] -public class DefaultBookmarkQueuePurger(IBookmarkQueueStore store, ISystemClock systemClock, ILogger logger) : IBookmarkQueuePurger +public class DefaultBookmarkQueuePurger(IBookmarkQueueStore store, ISystemClock systemClock, IOptions options, ILogger logger) : IBookmarkQueuePurger { - private readonly TimeSpan _ttl = TimeSpan.FromMinutes(1); - private readonly int _batchSize = 50; - public async Task PurgeAsync(CancellationToken cancellationToken = default) { var currentPage = 0; var now = systemClock.UtcNow; - var thresholdDate = now - _ttl; + var thresholdDate = now - options.Value.Ttl; logger.LogInformation("Purging bookmark queue items older than {ThresholdDate}.", thresholdDate); while (true) { - var pageArgs = PageArgs.FromPage(currentPage, _batchSize); + var pageArgs = PageArgs.FromPage(currentPage, options.Value.BatchSize); var filter = new BookmarkQueueFilter { CreatedAtLessThan = thresholdDate diff --git a/src/modules/Elsa.Workflows.Runtime/Stimuli/ExecuteWorkflowStimulus.cs b/src/modules/Elsa.Workflows.Runtime/Stimuli/ExecuteWorkflowStimulus.cs new file mode 100644 index 000000000..61a20a528 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Stimuli/ExecuteWorkflowStimulus.cs @@ -0,0 +1,9 @@ +using Elsa.Workflows.Runtime.Activities; + +namespace Elsa.Workflows.Runtime.Stimuli; + +/// +/// Bookmark payload for the activity. +/// +/// The instance ID of the child workflow that was created by the activity. +public record ExecuteWorkflowStimulus(string ChildInstanceId); \ No newline at end of file From f0615337ee2d7950c94ad348e6c43edb83e43cf8 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 26 Nov 2024 20:37:07 +0100 Subject: [PATCH 2/3] Update activity reference in XML comments Corrected the activity reference from DispatchWorkflow to ExecuteWorkflow in XML comments to accurately reflect the functionality. This change ensures the comments are aligned with the actual code behavior. --- .../Handlers/ResumeExecuteWorkflowActivity.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs index 9d4912d38..256621e0b 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs @@ -5,12 +5,11 @@ using Elsa.Workflows.Runtime.Activities; using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Stimuli; using JetBrains.Annotations; -using Microsoft.Extensions.Logging; namespace Elsa.Workflows.Runtime.Handlers; /// -/// Resumes any blocking activities when its child workflow completes. +/// Resumes any blocking activities when its child workflow completes. /// [PublicAPI] internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStimulusHasher stimulusHasher) : INotificationHandler From c5ce511e8bd95ee5d448297d032d56a5e7d9f9c5 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 26 Nov 2024 21:21:26 +0100 Subject: [PATCH 3/3] Switch to Distributed runtime and MassTransit caching. Updated the workflow runtime to Distributed and the distributed caching transport to MassTransit for better scalability and performance. Also, disabled the use of secrets to enhance security and simplify configuration. --- src/apps/Elsa.Server.Web/Program.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/apps/Elsa.Server.Web/Program.cs b/src/apps/Elsa.Server.Web/Program.cs index ecb451afd..6c80b6fce 100644 --- a/src/apps/Elsa.Server.Web/Program.cs +++ b/src/apps/Elsa.Server.Web/Program.cs @@ -68,12 +68,12 @@ const bool useAzureServiceBus = false; const bool useKafka = false; const bool useReadOnlyMode = false; const bool useSignalR = false; // Disabled until Elsa Studio sends authenticated requests. -const WorkflowRuntime workflowRuntime = WorkflowRuntime.ProtoActor; -const DistributedCachingTransport distributedCachingTransport = DistributedCachingTransport.ProtoActor; +const WorkflowRuntime workflowRuntime = WorkflowRuntime.Distributed; +const DistributedCachingTransport distributedCachingTransport = DistributedCachingTransport.MassTransit; const MassTransitBroker massTransitBroker = MassTransitBroker.Memory; const bool useMultitenancy = false; const bool useAgents = false; -const bool useSecrets = true; +const bool useSecrets = false; const bool disableVariableWrappers = false; var builder = WebApplication.CreateBuilder(args);