Add workflow restart functionality for handling interruptions

Introduced a mechanism to identify and restart interrupted workflows. This includes a new `IWorkflowRestarter` contract, its default implementation, and a recurring task for handling restarts. Additionally, updated configurations and added extensions to improve workflow instance filtering and liveness tracking.

Elsa.sln

@@ -881,7 +881,6 @@ Global
 		{C9539BD8-D2AE-4A8D-8281-71A05B3FBF31} = {B08B4E00-C2AB-48F3-8389-449F42AEF179}
 		{169E2C9B-6687-427F-A278-30BF849BFEDC} = {B08B4E00-C2AB-48F3-8389-449F42AEF179}
 		{2DD5D66B-85E9-4AF9-911C-C9F963234159} = {B08B4E00-C2AB-48F3-8389-449F42AEF179}
-		{8A050229-DB79-4E0B-9AFF-7565E87F2954} = {58C59255-281C-4595-8732-808158F1DC6E}
 		{C237BA1A-3A7D-4AB2-BE09-2696F3C082A4} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
 		{80529478-A383-4FEA-B744-C71264969E9A} = {6EF07978-A6D2-40EB-891D-7D70C5F37E76}
 		{39CD855E-83B1-4A96-93F7-01608211EBE3} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09}
@@ -1011,6 +1010,7 @@ Global
 		{FA5E857F-B173-4B5D-8049-B817A210DEF5} = {A0DC5F8E-5D7F-4E8A-A5DF-B1FC31F7336E}
 		{A51F9683-DA9F-45E7-82DE-1E261ACD6D68} = {A0DC5F8E-5D7F-4E8A-A5DF-B1FC31F7336E}
 		{66E2E2CF-967F-4564-89E8-F46FA973C99B} = {986E5482-0482-448C-B9E4-EC67A9474B85}
+		{8A050229-DB79-4E0B-9AFF-7565E87F2954} = {B08B4E00-C2AB-48F3-8389-449F42AEF179}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {D4B5CEAA-7D70-4FCB-A68E-B03FBE5E0E5E}

src/apps/Elsa.Server.Web/Program.cs

@@ -681,8 +681,10 @@ services.Configure<RecurringTaskOptions>(options =>
     options.Schedule.ConfigureTask<TriggerBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(300));
     options.Schedule.ConfigureTask<PurgeBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(300));
     options.Schedule.ConfigureTask<UpdateExpiredSecretsRecurringTask>(TimeSpan.FromHours(4));
+    options.Schedule.ConfigureTask<RestartInterruptedWorkflowsTask>(TimeSpan.FromSeconds(15));
 });
 
+services.Configure<RuntimeOptions>(options => { options.WorkflowLivenessThreshold = TimeSpan.FromSeconds(15); });
 services.Configure<BookmarkQueuePurgeOptions>(options => options.Ttl = TimeSpan.FromSeconds(10));
 
 services.Configure<CachingOptions>(options => options.CacheDuration = TimeSpan.FromDays(1));

src/apps/Elsa.Server.Web/SlowActivity.cs

@@ -0,0 +1,13 @@
+using Elsa.Workflows;
+
+namespace Elsa.Server.Web;
+
+public class SlowActivity : CodeActivity
+{
+    protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
+    {
+        Console.WriteLine("Starting...");
+        await Task.Delay(TimeSpan.FromMinutes(1));
+        Console.WriteLine("Done.");
+    }
+}

src/apps/Elsa.Server.Web/appsettings.json

@@ -2,7 +2,8 @@
   "Logging": {
     "LogLevel": {
       "Default": "Warning",
-      "Microsoft.Hosting.Lifetime": "Information"
+      "Microsoft.Hosting.Lifetime": "Information",
+      "Elsa": "Information"
     }
   },
   "HostBuilder": {

src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs

@@ -54,7 +54,6 @@ public class MassTransitWorkflowDispatcher(
     public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = null, CancellationToken cancellationToken = default)
     {
         var sendEndpoint = await GetSendEndpointAsync(options);
-        var serializedInput = SerializeInput(request.Input);
 
         await sendEndpoint.Send(new DispatchWorkflowInstance(request.InstanceId)
         {

src/modules/Elsa.Retention/Tasks/CleanupRecurringTask.cs

@@ -12,6 +12,7 @@ namespace Elsa.Retention;
 [UsedImplicitly]
 public class CleanupRecurringTask(CleanupJob job) : RecurringTask
 {
+    /// <inheritdoc />
     public override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         await job.ExecuteAsync(stoppingToken);

src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs

@@ -157,8 +157,20 @@ public class WorkflowRunner(
         }
         else
         {
-            // Nothing was scheduled. Schedule the workflow itself.
-            workflowExecutionContext.ScheduleWorkflow();
+            // Check if there are any leaf nodes in the Pending state.
+            var pendingActivityExecutionContexts = workflowExecutionContext.ActivityExecutionContexts.Where(x => x.Status == ActivityStatus.Pending).ToList();
+
+            if( pendingActivityExecutionContexts.Count > 0)
+            {
+                // Schedule the pending activities.
+                foreach (var pendingActivityExecutionContext in pendingActivityExecutionContexts)
+                    workflowExecutionContext.ScheduleActivityExecutionContext(pendingActivityExecutionContext);
+            }
+            else
+            {
+                // Nothing was scheduled. Schedule the workflow itself.
+                workflowExecutionContext.ScheduleWorkflow();
+            }
         }
 
         return await RunAsync(workflowExecutionContext);

src/modules/Elsa.Workflows.Management/Filters/WorkflowInstanceFilter.cs

@@ -101,6 +101,11 @@ public class WorkflowInstanceFilter
     /// </summary>
     public bool? IsSystem { get; set; }
 
+    /// <summary>
+    /// Filter workflow instances that are older than the specified timestamp.
+    /// </summary>
+    public DateTimeOffset? BeforeLastUpdated { get; set; }
+
     /// <summary>
     /// Filter workflow instances by timestamp.
     /// </summary>
@@ -131,6 +136,7 @@ public class WorkflowInstanceFilter
         if (filter.HasIncidents != null) query = filter.HasIncidents == true ? query.Where(x => x.IncidentCount > 0) : query.Where(x => x.IncidentCount == 0);
         if (filter.IsSystem != null) query = query.Where(x => x.IsSystem == filter.IsSystem);
         if (filter.Name != null) query = query.Where(x => x.Name!.ToLower().Contains(filter.Name.ToLower()));
+        if (filter.BeforeLastUpdated != null) query = query.Where(x => x.UpdatedAt < filter.BeforeLastUpdated);
 
         if (TimestampFilters != null)
         {

src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRestarter.cs

@@ -0,0 +1,19 @@
+using Elsa.Workflows.Runtime.Tasks;
+
+namespace Elsa.Workflows.Runtime;
+
+/// <summary>
+/// Defines the contract for restarting workflows in the runtime environment.
+/// </summary>
+/// <remarks>
+/// This service is used by the <see cref="RestartInterruptedWorkflowsTask"/> responsible for restarting interrupted workflows.
+/// </remarks>
+public interface IWorkflowRestarter
+{
+    /// <summary>
+    /// Restarts a workflow with the specified workflow instance ID.
+    /// </summary>
+    /// <param name="workflowInstanceId">The ID of the workflow instance to restart.</param>
+    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+    public Task RestartWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default);
+}

src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowStarter.cs

@@ -1,5 +1,9 @@
 namespace Elsa.Workflows.Runtime;
 
+/// <summary>
+/// Represents an interface responsible for starting workflows.
+/// Provides a method to start a workflow based on the provided request.
+/// </summary>
 public interface IWorkflowStarter
 {
     public Task<StartWorkflowResponse> StartWorkflowAsync(StartWorkflowRequest request, CancellationToken cancellationToken = default);

src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowInstanceStoreExtensions.cs

@@ -0,0 +1,33 @@
+using System.Runtime.CompilerServices;
+using Elsa.Common.Models;
+using Elsa.Workflows.Management;
+using Elsa.Workflows.Management.Filters;
+using Elsa.Workflows.Management.Models;
+
+namespace Elsa.Workflows.Runtime;
+
+public static class WorkflowInstanceStoreExtensions
+{
+    public static async IAsyncEnumerable<WorkflowInstanceSummary> EnumerateSummariesAsync(
+        this IWorkflowInstanceStore store, 
+        WorkflowInstanceFilter filter, 
+        int batchSize = 100, 
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        var pageArgs = PageArgs.FromPage(0, batchSize);
+
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            var page = await store.SummarizeManyAsync(filter, pageArgs, cancellationToken);
+            var workflowInstances = page.Items;
+
+            if (workflowInstances.Count == 0)
+                yield break;
+
+            foreach (var workflowInstance in workflowInstances)
+                yield return workflowInstance;
+
+            pageArgs = pageArgs.Next();
+        }
+    }
+}

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

@@ -262,6 +262,7 @@ public class WorkflowRuntimeFeature : FeatureBase
             .AddScoped<IWorkflowCancellationService, WorkflowCancellationService>()
             .AddScoped<IWorkflowActivationStrategyEvaluator, DefaultWorkflowActivationStrategyEvaluator>()
             .AddScoped<IWorkflowStarter, DefaultWorkflowStarter>()
+            .AddScoped<IWorkflowRestarter, DefaultWorkflowRestarter>()
             .AddScoped<IBookmarkQueuePurger, DefaultBookmarkQueuePurger>()
             .AddScoped<ILogRecordExtractor<WorkflowExecutionLogRecord>, WorkflowExecutionLogRecordExtractor>()
             
@@ -297,6 +298,7 @@ public class WorkflowRuntimeFeature : FeatureBase
             .AddStartupTask<PopulateRegistriesStartupTask>()
             .AddRecurringTask<TriggerBookmarkQueueRecurringTask>(TimeSpan.FromMinutes(1))
             .AddRecurringTask<PurgeBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(10))
+            .AddRecurringTask<RestartInterruptedWorkflowsTask>(TimeSpan.FromMinutes(5)) // Same default as the workflow liveness threshold.
 
             // Distributed locking.
             .AddSingleton(DistributedLockProvider)

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

@@ -1,3 +1,5 @@
+using Elsa.Workflows.Management.Entities;
+
 namespace Elsa.Workflows.Runtime.Options;
 
 /// <summary>
@@ -9,4 +11,24 @@ public class RuntimeOptions
     /// A list of workflow builders configured during application startup.
     /// </summary>
     public IDictionary<string, Func<IServiceProvider, ValueTask<IWorkflow>>> Workflows { get; set; } = new Dictionary<string, Func<IServiceProvider, ValueTask<IWorkflow>>>();
+
+    /// <summary>
+    /// The default workflow liveness threshold.
+    /// </summary>
+    /// <remarks>
+    /// The liveness threshold is used to determine if a persisted workflow instance in the <see cref="WorkflowSubStatus.Executing"/> state should be considered interrupted or not.
+    /// Interrupted workflows will be attempted to be restarted.
+    /// A separate heartbeat process will ensure the <see cref="WorkflowInstance.UpdatedAt"/> is updated before this threshold.
+    /// If the workflow instance got removed from memory, e.g. because of an application shutdown, the LastUpdated field will eventually exceed the liveness threshold and therefore be considered to be interrupted.
+    /// </remarks>
+    public TimeSpan WorkflowLivenessThreshold { get; set; } = TimeSpan.FromMinutes(5);
+
+    /// <summary>
+    /// The number of workflow instances to restart in a single batch.
+    /// </summary>
+    /// <remarks>
+    /// The batch size represents the number of workflow instance records to load into memory at a time.
+    /// This provides control over memory consumption of the application.
+    /// </remarks>
+    public int RestartInterruptedWorkflowsBatchSize { get; set; } = 100;
 }

src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceRequest.cs

@@ -28,7 +28,7 @@ public class DispatchWorkflowInstanceRequest
     /// <summary>
     /// The ID of the workflow instance to dispatch.
     /// </summary>
-    public string InstanceId { get; init; } = default!;
+    public string InstanceId { get; init; } = null!;
     
     /// <summary>
     /// The ID of the bookmark to resume.

src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRestarter.cs

@@ -0,0 +1,18 @@
+using Elsa.Workflows.Runtime.Requests;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Workflows.Runtime;
+
+/// <inheritdoc />
+public class DefaultWorkflowRestarter(IWorkflowDispatcher workflowDispatcher, ILogger<DefaultWorkflowRestarter> logger) : IWorkflowRestarter
+{
+    /// <inheritdoc />
+    public async Task RestartWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
+    {
+        var request = new DispatchWorkflowInstanceRequest(workflowInstanceId);
+        var options = new DispatchWorkflowOptions();
+        
+        logger.LogInformation("Restarting workflow {WorkflowInstanceId}", workflowInstanceId);
+        await workflowDispatcher.DispatchAsync(request, options, cancellationToken);
+    }
+}

src/modules/Elsa.Workflows.Runtime/Tasks/RestartInterruptedWorkflowsTask.cs

@@ -0,0 +1,47 @@
+using Elsa.Common;
+using Elsa.Common.RecurringTasks;
+using Elsa.Workflows.Management;
+using Elsa.Workflows.Management.Filters;
+using Elsa.Workflows.Runtime.Options;
+using JetBrains.Annotations;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Elsa.Workflows.Runtime.Tasks;
+
+[SingleNodeTask]
+[UsedImplicitly]
+public class RestartInterruptedWorkflowsTask(
+    IWorkflowInstanceStore workflowInstanceStore, 
+    IWorkflowRestarter workflowRestarter, 
+    IOptions<RuntimeOptions> options, 
+    ISystemClock systemClock,
+    ILogger<RestartInterruptedWorkflowsTask> logger) : RecurringTask
+{
+    /// <inheritdoc />
+    public override async Task ExecuteAsync(CancellationToken cancellationToken)
+    {
+        var workflowInstanceFilter = CreateWorkflowInstanceFilter();
+        var batchSize = options.Value.RestartInterruptedWorkflowsBatchSize;
+        var workflowInstances = workflowInstanceStore.EnumerateSummariesAsync(workflowInstanceFilter, batchSize, cancellationToken);
+
+        logger.LogInformation("Restarting interrupted workflows.");
+        await foreach (var workflowInstance in workflowInstances)
+        {
+            await workflowRestarter.RestartWorkflowAsync(workflowInstance.Id, cancellationToken: cancellationToken);
+        }
+        logger.LogInformation("Finished restarting interrupted workflows.");
+    }
+
+    private WorkflowInstanceFilter CreateWorkflowInstanceFilter()
+    {
+        var livenessThreshold = options.Value.WorkflowLivenessThreshold;
+        var now = systemClock.UtcNow;
+        var cutoffTimestamp = now - livenessThreshold;
+        return new()
+        {
+            WorkflowSubStatus = WorkflowSubStatus.Executing,
+            BeforeLastUpdated = cutoffTimestamp
+        };
+    }
+}