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

@@ -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<RecurringTaskOptions>(options =>
 {
     options.Schedule.ConfigureTask<TriggerBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(30));
     options.Schedule.ConfigureTask<UpdateExpiredSecretsRecurringTask>(TimeSpan.FromHours(4));
-    options.Schedule.ConfigureTask<PurgeBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(60));
+    options.Schedule.ConfigureTask<PurgeBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(11));
 });
 
+services.Configure<BookmarkQueuePurgeOptions>(options => options.Ttl = TimeSpan.FromSeconds(10));
+
 //services.Configure<CachingOptions>(options => options.CacheDuration = TimeSpan.FromDays(1));
 services.AddHealthChecks();
 services.AddControllers();

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<ExecuteWorkflowResult>
     /// </summary>
     [Input(Description = "The input to send to the workflow.")]
     public Input<IDictionary<string, object>?> Input { get; set; } = default!;
+    
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [Input(Description = "Wait for the child workflow to complete before completing this activity.")]
+    public Input<bool> WaitForCompletion { get; set; } = default!;
 
     /// <inheritdoc />
     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<ExecuteWorkflowResult> ExecuteWorkflowAsync(ActivityExecutionContext context)
@@ -87,4 +109,11 @@ public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
 
         return info;
     }
+    
+    private async ValueTask OnChildWorkflowCompletedAsync(ActivityExecutionContext context)
+    {
+        var input = context.WorkflowInput;
+        context.Set(Result, input);
+        await context.CompleteActivityAsync();
+    }
 }

src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs

@@ -137,6 +137,11 @@ public class WorkflowRuntimeFeature : FeatureBase
     /// A delegate to configure the <see cref="WorkflowDispatcherOptions"/>.
     /// </summary>
     public Action<WorkflowDispatcherOptions> WorkflowDispatcherOptions { get; set; } = _ => { };
+    
+    /// <summary>
+    /// A delegate to configure the <see cref="BookmarkQueuePurgeOptions"/>.
+    /// </summary>
+    public Action<BookmarkQueuePurgeOptions> BookmarkQueuePurgeOptions { get; set; } = _ => { };
 
     /// <summary>
     /// 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<RuntimeOptions>(options => { options.Workflows = Workflows; });
         Services.Configure<WorkflowDispatcherOptions>(options =>
         {
@@ -292,7 +298,7 @@ public class WorkflowRuntimeFeature : FeatureBase
             // Startup tasks, background tasks, and recurring tasks.
             .AddStartupTask<PopulateRegistriesStartupTask>()
             .AddRecurringTask<TriggerBookmarkQueueRecurringTask>(TimeSpan.FromMinutes(1))
-            .AddRecurringTask<PurgeBookmarkQueueRecurringTask>(TimeSpan.FromMinutes(1))
+            .AddRecurringTask<PurgeBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(10))
 
             // Distributed locking.
             .AddSingleton(DistributedLockProvider)
@@ -300,13 +306,14 @@ public class WorkflowRuntimeFeature : FeatureBase
             // Workflow definition providers.
             .AddWorkflowDefinitionProvider<ClrWorkflowsProvider>()
             
-            // UI prooprty handlers.
+            // UI property handlers.
             .AddScoped<IPropertyUIHandler, DispatcherChannelOptionsProvider>()
 
             // Domain handlers.
             .AddCommandHandler<DispatchWorkflowCommandHandler>()
             .AddNotificationHandler<ResumeDispatchWorkflowActivity>()
             .AddNotificationHandler<ResumeBulkDispatchWorkflowActivity>()
+            .AddNotificationHandler<ResumeExecuteWorkflowActivity>()
             .AddNotificationHandler<IndexTriggers>()
             .AddNotificationHandler<CancelBackgroundActivities>()
             .AddNotificationHandler<DeleteBookmarks>()

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;
+
+/// <summary>
+/// Resumes any blocking <see cref="DispatchWorkflow"/> activities when its child workflow completes.
+/// </summary>
+[PublicAPI]
+internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStimulusHasher stimulusHasher) : INotificationHandler<WorkflowExecuted>
+{
+    private static readonly string ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<ExecuteWorkflow>();
+
+    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);
+    }
+}

src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs

@@ -0,0 +1,17 @@
+namespace Elsa.Workflows.Runtime.Options;
+
+/// <summary>
+/// Options for purging the bookmark queue.
+/// </summary>
+public class BookmarkQueuePurgeOptions
+{
+    /// <summary>
+    /// The time-to-live for bookmark queue items.
+    /// </summary>
+    public TimeSpan Ttl { get; set; } = TimeSpan.FromSeconds(10);
+
+    /// <summary>
+    /// The number of records to clean up per sweep.
+    /// </summary>
+    public int BatchSize { get; set; } = 1000;
+}

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<DefaultBookmarkQueuePurger> logger) : IBookmarkQueuePurger
+public class DefaultBookmarkQueuePurger(IBookmarkQueueStore store, ISystemClock systemClock, IOptions<BookmarkQueuePurgeOptions> options, ILogger<DefaultBookmarkQueuePurger> 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

src/modules/Elsa.Workflows.Runtime/Stimuli/ExecuteWorkflowStimulus.cs

@@ -0,0 +1,9 @@
+using Elsa.Workflows.Runtime.Activities;
+
+namespace Elsa.Workflows.Runtime.Stimuli;
+
+/// <summary>
+/// Bookmark payload for the <see cref="ExecuteWorkflow"/> activity.
+/// </summary>
+/// <param name="ChildInstanceId">The instance ID of the child workflow that was created by the <see cref="ExecuteWorkflow"/> activity.</param>
+public record ExecuteWorkflowStimulus(string ChildInstanceId);