diff --git a/Elsa.sln b/Elsa.sln index 3bd4eac22..bd60f8f71 100644 --- a/Elsa.sln +++ b/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} diff --git a/src/apps/Elsa.Server.Web/Program.cs b/src/apps/Elsa.Server.Web/Program.cs index aeceb8079..0816171cd 100644 --- a/src/apps/Elsa.Server.Web/Program.cs +++ b/src/apps/Elsa.Server.Web/Program.cs @@ -681,8 +681,10 @@ services.Configure(options => options.Schedule.ConfigureTask(TimeSpan.FromSeconds(300)); options.Schedule.ConfigureTask(TimeSpan.FromSeconds(300)); options.Schedule.ConfigureTask(TimeSpan.FromHours(4)); + options.Schedule.ConfigureTask(TimeSpan.FromSeconds(15)); }); +services.Configure(options => { options.WorkflowLivenessThreshold = TimeSpan.FromSeconds(15); }); services.Configure(options => options.Ttl = TimeSpan.FromSeconds(10)); services.Configure(options => options.CacheDuration = TimeSpan.FromDays(1)); diff --git a/src/apps/Elsa.Server.Web/SlowActivity.cs b/src/apps/Elsa.Server.Web/SlowActivity.cs new file mode 100644 index 000000000..633fb2f00 --- /dev/null +++ b/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."); + } +} \ No newline at end of file diff --git a/src/apps/Elsa.Server.Web/appsettings.json b/src/apps/Elsa.Server.Web/appsettings.json index a1d35303b..a43954d17 100644 --- a/src/apps/Elsa.Server.Web/appsettings.json +++ b/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": { diff --git a/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs index e7b7d64ce..fde8913f0 100644 --- a/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs +++ b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs @@ -54,7 +54,6 @@ public class MassTransitWorkflowDispatcher( public async Task 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) { diff --git a/src/modules/Elsa.Retention/Tasks/CleanupRecurringTask.cs b/src/modules/Elsa.Retention/Tasks/CleanupRecurringTask.cs index 8f39eff09..49fcf3ac7 100644 --- a/src/modules/Elsa.Retention/Tasks/CleanupRecurringTask.cs +++ b/src/modules/Elsa.Retention/Tasks/CleanupRecurringTask.cs @@ -12,6 +12,7 @@ namespace Elsa.Retention; [UsedImplicitly] public class CleanupRecurringTask(CleanupJob job) : RecurringTask { + /// public override async Task ExecuteAsync(CancellationToken stoppingToken) { await job.ExecuteAsync(stoppingToken); diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs index a79842799..a6a0edc11 100644 --- a/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs +++ b/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); diff --git a/src/modules/Elsa.Workflows.Management/Filters/WorkflowInstanceFilter.cs b/src/modules/Elsa.Workflows.Management/Filters/WorkflowInstanceFilter.cs index 15deb1e41..b56e04915 100644 --- a/src/modules/Elsa.Workflows.Management/Filters/WorkflowInstanceFilter.cs +++ b/src/modules/Elsa.Workflows.Management/Filters/WorkflowInstanceFilter.cs @@ -101,6 +101,11 @@ public class WorkflowInstanceFilter /// public bool? IsSystem { get; set; } + /// + /// Filter workflow instances that are older than the specified timestamp. + /// + public DateTimeOffset? BeforeLastUpdated { get; set; } + /// /// Filter workflow instances by timestamp. /// @@ -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) { diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRestarter.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRestarter.cs new file mode 100644 index 000000000..bb9be1292 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRestarter.cs @@ -0,0 +1,19 @@ +using Elsa.Workflows.Runtime.Tasks; + +namespace Elsa.Workflows.Runtime; + +/// +/// Defines the contract for restarting workflows in the runtime environment. +/// +/// +/// This service is used by the responsible for restarting interrupted workflows. +/// +public interface IWorkflowRestarter +{ + /// + /// Restarts a workflow with the specified workflow instance ID. + /// + /// The ID of the workflow instance to restart. + /// A token to monitor for cancellation requests. + public Task RestartWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowStarter.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowStarter.cs index cb4c17062..68de70e37 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowStarter.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowStarter.cs @@ -1,5 +1,9 @@ namespace Elsa.Workflows.Runtime; +/// +/// Represents an interface responsible for starting workflows. +/// Provides a method to start a workflow based on the provided request. +/// public interface IWorkflowStarter { public Task StartWorkflowAsync(StartWorkflowRequest request, CancellationToken cancellationToken = default); diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowInstanceStoreExtensions.cs b/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowInstanceStoreExtensions.cs new file mode 100644 index 000000000..32616dbe7 --- /dev/null +++ b/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 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(); + } + } +} \ 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 5ee379d74..16b1a5543 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -262,6 +262,7 @@ public class WorkflowRuntimeFeature : FeatureBase .AddScoped() .AddScoped() .AddScoped() + .AddScoped() .AddScoped() .AddScoped, WorkflowExecutionLogRecordExtractor>() @@ -297,6 +298,7 @@ public class WorkflowRuntimeFeature : FeatureBase .AddStartupTask() .AddRecurringTask(TimeSpan.FromMinutes(1)) .AddRecurringTask(TimeSpan.FromSeconds(10)) + .AddRecurringTask(TimeSpan.FromMinutes(5)) // Same default as the workflow liveness threshold. // Distributed locking. .AddSingleton(DistributedLockProvider) diff --git a/src/modules/Elsa.Workflows.Runtime/Options/RuntimeOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/RuntimeOptions.cs index 97aab8d47..1eb7fcebf 100644 --- a/src/modules/Elsa.Workflows.Runtime/Options/RuntimeOptions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Options/RuntimeOptions.cs @@ -1,3 +1,5 @@ +using Elsa.Workflows.Management.Entities; + namespace Elsa.Workflows.Runtime.Options; /// @@ -9,4 +11,24 @@ public class RuntimeOptions /// A list of workflow builders configured during application startup. /// public IDictionary>> Workflows { get; set; } = new Dictionary>>(); + + /// + /// The default workflow liveness threshold. + /// + /// + /// The liveness threshold is used to determine if a persisted workflow instance in the state should be considered interrupted or not. + /// Interrupted workflows will be attempted to be restarted. + /// A separate heartbeat process will ensure the 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. + /// + public TimeSpan WorkflowLivenessThreshold { get; set; } = TimeSpan.FromMinutes(5); + + /// + /// The number of workflow instances to restart in a single batch. + /// + /// + /// 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. + /// + public int RestartInterruptedWorkflowsBatchSize { get; set; } = 100; } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceRequest.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceRequest.cs index bffcfa9c7..65ad250b1 100644 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceRequest.cs +++ b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceRequest.cs @@ -28,7 +28,7 @@ public class DispatchWorkflowInstanceRequest /// /// The ID of the workflow instance to dispatch. /// - public string InstanceId { get; init; } = default!; + public string InstanceId { get; init; } = null!; /// /// The ID of the bookmark to resume. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRestarter.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRestarter.cs new file mode 100644 index 000000000..88165636e --- /dev/null +++ b/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; + +/// +public class DefaultWorkflowRestarter(IWorkflowDispatcher workflowDispatcher, ILogger logger) : IWorkflowRestarter +{ + /// + 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); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Tasks/RestartInterruptedWorkflowsTask.cs b/src/modules/Elsa.Workflows.Runtime/Tasks/RestartInterruptedWorkflowsTask.cs new file mode 100644 index 000000000..7f5191e46 --- /dev/null +++ b/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 options, + ISystemClock systemClock, + ILogger logger) : RecurringTask +{ + /// + 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 + }; + } +} \ No newline at end of file