Make Quartz.NET Jobs Resilient

.github/workflows/packages.yml

@@ -13,7 +13,7 @@ on:
   release:
     types: [ prereleased, published ]
 env:
-  base_version: '3.6.0'
+  base_version: '3.7.0'
   feedz_feed_source: 'https://f.feedz.io/elsa-workflows/elsa-3/nuget/index.json'
   nuget_feed_source: 'https://api.nuget.org/v3/index.json'
 

Directory.Build.props

@@ -37,7 +37,7 @@
 	</PropertyGroup>
 
     <PropertyGroup Label="PackageVersions">
-        <ElsaVersion>3.6.0-rc1</ElsaVersion>
+        <ElsaVersion>3.7.0-preview.4150</ElsaVersion>
         <ElsaStudioVersion>3.6.0-rc1</ElsaStudioVersion>
     </PropertyGroup>
 

Directory.Packages.props

@@ -28,6 +28,7 @@
         <PackageVersion Include="Elsa.Persistence.EFCore.Common" Version="$(ElsaVersion)"/>
         <PackageVersion Include="Elsa.Persistence.EFCore" Version="$(ElsaVersion)"/>
         <PackageVersion Include="Elsa.Resilience" Version="$(ElsaVersion)"/>
+        <PackageVersion Include="Elsa.Resilience.Core" Version="$(ElsaVersion)"/>
         <PackageVersion Include="Elsa.Tenants" Version="$(ElsaVersion)"/>
         <PackageVersion Include="Elsa.Tenants.AspNetCore" Version="$(ElsaVersion)"/>
         <PackageVersion Include="Elsa.Testing.Shared" Version="$(ElsaVersion)"/>

src/modules/scheduling/Elsa.Scheduling.Quartz.EFCore.MySql/MySqlQuartzExtensions.cs

@@ -35,7 +35,7 @@ public static QuartzFeature UseMySql(this QuartzFeature feature, string connecti
                     store.UseClustering();
             });
         };
-        
+
         feature.Module.ConfigureHostedService<RunMigrationsHostedService<MySqlQuartzDbContext>>(-100);
 
         return feature;

src/modules/scheduling/Elsa.Scheduling.Quartz/Contracts/IQuartzJobRetryScheduler.cs

@@ -0,0 +1,14 @@
+using Quartz;
+
+namespace Elsa.Scheduling.Quartz.Contracts;
+
+/// <summary>
+/// Schedules retries for Quartz jobs.
+/// </summary>
+public interface IQuartzJobRetryScheduler
+{
+    /// <summary>
+    /// Reschedules the trigger of the current job for a retry using the configured retry delay.
+    /// </summary>
+    Task ScheduleRetryAsync(IJobExecutionContext context, CancellationToken cancellationToken = default);
+}

src/modules/scheduling/Elsa.Scheduling.Quartz/Elsa.Scheduling.Quartz.csproj

@@ -13,10 +13,12 @@
 
     <ItemGroup Label="Elsa" Condition="'$(UseProjectReferences)' != 'true'">
         <PackageReference Include="Elsa.Scheduling"/>
+        <PackageReference Include="Elsa.Resilience.Core"/>
     </ItemGroup>
 
     <ItemGroup Label="Elsa" Condition="'$(UseProjectReferences)' == 'true'">
         <ProjectReference Include="..\..\..\..\..\elsa-core\src\modules\Elsa.Scheduling\Elsa.Scheduling.csproj"/>
+        <ProjectReference Include="..\..\..\..\..\elsa-core\src\modules\Elsa.Resilience.Core\Elsa.Resilience.Core.csproj"/>
     </ItemGroup>
 
 </Project>

src/modules/scheduling/Elsa.Scheduling.Quartz/Features/QuartzSchedulerFeature.cs

@@ -2,12 +2,14 @@
 using Elsa.Features.Abstractions;
 using Elsa.Features.Attributes;
 using Elsa.Features.Services;
-using Elsa.Scheduling.Quartz.Contracts;
 using Elsa.Scheduling.Quartz.Handlers;
+using Elsa.Scheduling.Quartz.Options;
 using Elsa.Scheduling.Quartz.Services;
 using Elsa.Scheduling.Quartz.Tasks;
 using Elsa.Scheduling.Features;
+using Elsa.Scheduling.Quartz.Contracts;
 using Elsa.Workflows;
+using JetBrains.Annotations;
 using Microsoft.Extensions.DependencyInjection;
 using Quartz;
 
@@ -17,8 +19,18 @@ namespace Elsa.Scheduling.Quartz.Features;
 /// A feature that installs Quartz.NET implementations for <see cref="IWorkflowScheduler"/>.
 /// </summary>
 [DependsOn(typeof(SchedulingFeature))]
+[UsedImplicitly]
 public class QuartzSchedulerFeature(IModule module) : FeatureBase(module)
 {
+    private Action<QuartzJobOptions> _configureQuartzJobOptions = _ => { };
+
+    [PublicAPI]
+    public QuartzSchedulerFeature ConfigureOptions(Action<QuartzJobOptions> configure)
+    {
+        _configureQuartzJobOptions += configure;
+        return this;
+    }
+
     /// <inheritdoc />
     public override void Configure()
     {
@@ -40,7 +52,10 @@ public override void Apply()
             .AddSingleton<QuartzCronParser>()
             .AddScoped<QuartzWorkflowScheduler>()
             .AddScoped<IJobKeyProvider, JobKeyProvider>()
+            .AddSingleton<IQuartzJobRetryScheduler, QuartzJobRetryScheduler>()
             .AddStartupTask<RegisterJobsTask>()
             .AddQuartz();
+
+        Services.Configure(_configureQuartzJobOptions);
     }
 }

src/modules/scheduling/Elsa.Scheduling.Quartz/Jobs/ResumeWorkflowJob.cs

@@ -1,17 +1,27 @@
 using Elsa.Common;
 using Elsa.Common.Multitenancy;
 using Elsa.Extensions;
+using Elsa.Resilience;
+using Elsa.Scheduling.Quartz.Contracts;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Messages;
+using Microsoft.Extensions.Logging;
 using Quartz;
 
 namespace Elsa.Scheduling.Quartz.Jobs;
 
 /// <summary>
 /// A job that resumes a workflow.
 /// </summary>
-public class ResumeWorkflowJob(IWorkflowRuntime workflowRuntime, IJsonSerializer jsonSerializer, ITenantFinder tenantFinder, ITenantAccessor tenantAccessor) : IJob
+public class ResumeWorkflowJob(
+    IWorkflowRuntime workflowRuntime,
+    IJsonSerializer jsonSerializer,
+    ITenantFinder tenantFinder,
+    ITenantAccessor tenantAccessor,
+    ITransientExceptionDetector transientExceptionDetector,
+    IQuartzJobRetryScheduler retryScheduler,
+    ILogger<ResumeWorkflowJob> logger) : IJob
 {
     /// <inheritdoc />
     public async Task Execute(IJobExecutionContext context)
@@ -23,15 +33,32 @@ public async Task Execute(IJobExecutionContext context)
             var serializedActivityHandle = (string)map.Get(nameof(ScheduleExistingWorkflowInstanceRequest.ActivityHandle));
             var activityHandle = serializedActivityHandle != null! ? jsonSerializer.Deserialize<ActivityHandle>(serializedActivityHandle) : null;
             var workflowInstanceId = (string)map.Get(nameof(ScheduleExistingWorkflowInstanceRequest.WorkflowInstanceId));
-            var workflowClient = await workflowRuntime.CreateClientAsync(workflowInstanceId, context.CancellationToken);
-            var request = new RunWorkflowInstanceRequest
+            var cancellationToken = context.CancellationToken;
+
+            try
+            {
+                var workflowClient = await workflowRuntime.CreateClientAsync(workflowInstanceId, cancellationToken);
+                var request = new RunWorkflowInstanceRequest
+                {
+                    BookmarkId = (string)map.Get(nameof(ScheduleExistingWorkflowInstanceRequest.BookmarkId)),
+                    ActivityHandle = activityHandle,
+                    Input = map.GetDictionary(nameof(ScheduleExistingWorkflowInstanceRequest.Input)),
+                    Properties = map.GetDictionary(nameof(ScheduleExistingWorkflowInstanceRequest.Properties)),
+                };
+                await workflowClient.RunInstanceAsync(request, cancellationToken: cancellationToken);
+
+                logger.LogInformation("Resumed workflow instance {WorkflowInstanceId}", workflowInstanceId);
+            }
+            catch (Exception e) when (transientExceptionDetector.IsTransient(e))
+            {
+                logger.LogWarning(e, "A transient error occurred while resuming workflow instance {WorkflowInstanceId}. Rescheduling job for retry", workflowInstanceId);
+                await retryScheduler.ScheduleRetryAsync(context, cancellationToken);
+            }
+            catch (Exception e)
             {
-                BookmarkId = (string)map.Get(nameof(ScheduleExistingWorkflowInstanceRequest.BookmarkId)),
-                ActivityHandle = activityHandle,
-                Input = map.GetDictionary(nameof(ScheduleExistingWorkflowInstanceRequest.Input)),
-                Properties = map.GetDictionary(nameof(ScheduleExistingWorkflowInstanceRequest.Properties)),
-            };
-            await workflowClient.RunInstanceAsync(request, cancellationToken: context.CancellationToken);
+                logger.LogError(e, "An error occurred while resuming workflow instance {WorkflowInstanceId}", workflowInstanceId);
+                await context.Scheduler.DeleteJob(context.JobDetail.Key, cancellationToken);
+            }
         }
     }
 }

src/modules/scheduling/Elsa.Scheduling.Quartz/Jobs/RunWorkflowJob.cs

@@ -1,8 +1,11 @@
 using Elsa.Common.Multitenancy;
 using Elsa.Extensions;
+using Elsa.Resilience;
+using Elsa.Scheduling.Quartz.Contracts;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Exceptions;
+using JetBrains.Annotations;
 using Microsoft.Extensions.Logging;
 using Quartz;
 
@@ -11,10 +14,13 @@ namespace Elsa.Scheduling.Quartz.Jobs;
 /// <summary>
 /// A job that runs a workflow.
 /// </summary>
+[UsedImplicitly]
 public class RunWorkflowJob(
     ITenantAccessor tenantAccessor,
     ITenantFinder tenantFinder,
     IWorkflowStarter workflowStarter,
+    ITransientExceptionDetector transientExceptionDetector,
+    IQuartzJobRetryScheduler retryScheduler,
     ILogger<RunWorkflowJob> logger) : IJob
 {
     /// <inheritdoc />
@@ -54,6 +60,11 @@ public async Task Execute(IJobExecutionContext context)
                 logger.LogWarning(e, "Could not find workflow graph for workflow definition handle {WorkflowDefinitionHandle}", startRequest.WorkflowDefinitionHandle);
                 await context.Scheduler.DeleteJob(context.JobDetail.Key, cancellationToken);
             }
+            catch (Exception e) when (transientExceptionDetector.IsTransient(e))
+            {
+                logger.LogWarning(e, "A transient error occurred while starting workflow {WorkflowDefinitionHandle} with correlation ID {CorrelationId}. Rescheduling job for retry", startRequest.WorkflowDefinitionHandle, startRequest.CorrelationId);
+                await retryScheduler.ScheduleRetryAsync(context, cancellationToken);
+            }
             catch (Exception e)
             {
                 logger.LogError(e, "An error occurred while starting workflow {WorkflowDefinitionHandle} with correlation ID {CorrelationId}", startRequest.WorkflowDefinitionHandle, startRequest.CorrelationId);

src/modules/scheduling/Elsa.Scheduling.Quartz/Options/QuartzJobOptions.cs

@@ -0,0 +1,12 @@
+namespace Elsa.Scheduling.Quartz.Options;
+
+/// <summary>
+/// Options for Quartz job execution behavior.
+/// </summary>
+public class QuartzJobOptions
+{
+    /// <summary>
+    /// The delay in seconds before rescheduling a job after a transient failure.
+    /// </summary>
+    public TimeSpan TransientExceptionRetryDelay { get; set; } = TimeSpan.FromSeconds(10);
+}

src/modules/scheduling/Elsa.Scheduling.Quartz/Services/QuartzJobRetryScheduler.cs

@@ -0,0 +1,28 @@
+using Elsa.Common;
+using Elsa.Scheduling.Quartz.Contracts;
+using Elsa.Scheduling.Quartz.Options;
+using Microsoft.Extensions.Options;
+using Quartz;
+
+namespace Elsa.Scheduling.Quartz.Services;
+
+/// <summary>
+/// Default implementation of <see cref="IQuartzJobRetryScheduler"/>.
+/// </summary>
+public class QuartzJobRetryScheduler(ISystemClock systemClock, IOptions<QuartzJobOptions> options) : IQuartzJobRetryScheduler
+{
+    /// <inheritdoc />
+    public async Task ScheduleRetryAsync(IJobExecutionContext context, CancellationToken cancellationToken = default)
+    {
+        var delay = options.Value.TransientExceptionRetryDelay;
+        var now = systemClock.UtcNow;
+
+        var trigger = TriggerBuilder.Create()
+            .ForJob(context.JobDetail.Key)
+            .StartAt(now.Add(delay))
+            .Build();
+
+        await context.Scheduler.RescheduleJob(context.Trigger.Key, trigger, cancellationToken);
+    }
+}
+

test/modules/scheduling/Elsa.Scheduling.Quartz.ComponentTests/Abstractions/AppComponentTest.cs

@@ -0,0 +1,22 @@
+using Elsa.Scheduling.Quartz.ComponentTests.Fixtures;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.Scheduling.Quartz.ComponentTests.Abstractions;
+
+[Collection(nameof(SchedulingAppCollection))]
+public abstract class AppComponentTest(SchedulingApp app) : IDisposable
+{
+    protected WorkflowServer WorkflowServer { get; } = app.WorkflowServer;
+    protected IServiceScope Scope { get; private set; } = app.WorkflowServer.Services.CreateScope();
+
+    public void Dispose()
+    {
+        OnDispose();
+        Scope?.Dispose();
+        GC.SuppressFinalize(this);
+    }
+
+    protected virtual void OnDispose()
+    {
+    }
+}

test/modules/scheduling/Elsa.Scheduling.Quartz.ComponentTests/Elsa.Scheduling.Quartz.ComponentTests.csproj

@@ -0,0 +1,13 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <ItemGroup>
+        <PackageReference Include="Elsa.Testing.Shared.Component" />
+        <PackageReference Include="Elsa.Workflows.Core" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\..\src\modules\scheduling\Elsa.Scheduling.Quartz\Elsa.Scheduling.Quartz.csproj" />
+        <ProjectReference Include="..\..\..\workbench\Elsa.TestServer.Web\Elsa.TestServer.Web.csproj" />
+    </ItemGroup>
+
+</Project>

test/modules/scheduling/Elsa.Scheduling.Quartz.ComponentTests/Fixtures/SchedulingApp.cs

@@ -0,0 +1,16 @@
+using JetBrains.Annotations;
+
+namespace Elsa.Scheduling.Quartz.ComponentTests.Fixtures;
+
+[UsedImplicitly]
+public class SchedulingApp : IAsyncLifetime
+{
+    public WorkflowServer WorkflowServer { get; } = new("http://localhost:5010");
+
+    public Task InitializeAsync() => Task.CompletedTask;
+
+    public async Task DisposeAsync()
+    {
+        await WorkflowServer.DisposeAsync();
+    }
+}

test/modules/scheduling/Elsa.Scheduling.Quartz.ComponentTests/Fixtures/SchedulingAppCollection.cs

@@ -0,0 +1,4 @@
+namespace Elsa.Scheduling.Quartz.ComponentTests.Fixtures;
+
+[CollectionDefinition(nameof(SchedulingAppCollection))]
+public class SchedulingAppCollection : ICollectionFixture<SchedulingApp>;

test/modules/scheduling/Elsa.Scheduling.Quartz.ComponentTests/Fixtures/WorkflowServer.cs

@@ -0,0 +1,38 @@
+using Elsa.Extensions;
+using Elsa.Scheduling.Quartz.Options;
+using JetBrains.Annotations;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Mvc.Testing;
+using Microsoft.AspNetCore.TestHost;
+using Microsoft.Extensions.DependencyInjection;
+using Quartz;
+
+namespace Elsa.Scheduling.Quartz.ComponentTests.Fixtures;
+
+[UsedImplicitly]
+public class WorkflowServer(string url) : WebApplicationFactory<Program>
+{
+    public async Task<global::Quartz.IScheduler> GetSchedulerAsync()
+    {
+        var factory = Services.GetRequiredService<ISchedulerFactory>();
+        return await factory.GetScheduler();
+    }
+
+    protected override void ConfigureWebHost(IWebHostBuilder builder)
+    {
+        builder.UseUrls(url);
+
+        TestServer.Web.Program.ConfigureForTest ??= elsa =>
+        {
+            elsa.UseQuartz();
+            elsa.UseScheduling(scheduling => scheduling.UseQuartzScheduler());
+            elsa.UseWorkflowRuntime();
+        };
+
+        builder.ConfigureTestServices(services =>
+        {
+            // Configure Quartz job options for fast retries in tests
+            services.Configure<QuartzJobOptions>(options => options.TransientExceptionRetryDelay = TimeSpan.FromSeconds(1));
+        });
+    }
+}

test/modules/scheduling/Elsa.Scheduling.Quartz.ComponentTests/Helpers/FailingWorkflowStarter.cs

@@ -0,0 +1,32 @@
+using Elsa.Workflows.Runtime;
+
+namespace Elsa.Scheduling.Quartz.ComponentTests.Helpers;
+
+/// <summary>
+/// A workflow starter that can be configured to fail with transient or non-transient exceptions.
+/// </summary>
+public class FailingWorkflowStarter(IWorkflowStarter innerStarter) : IWorkflowStarter
+{
+    private int _callCount;
+
+    public int FailuresBeforeSuccess { get; set; }
+    public Exception? ExceptionToThrow { get; set; }
+    public int CallCount => _callCount;
+
+    public async Task<StartWorkflowResponse> StartWorkflowAsync(StartWorkflowRequest request, CancellationToken cancellationToken = default)
+    {
+        _callCount++;
+
+        if (ExceptionToThrow != null && _callCount <= FailuresBeforeSuccess)
+        {
+            throw ExceptionToThrow;
+        }
+
+        return await innerStarter.StartWorkflowAsync(request, cancellationToken);
+    }
+
+    public void Reset()
+    {
+        _callCount = 0;
+    }
+}