Harden Quartz.NET integration for clustered deployments

src/modules/scheduling/Elsa.Scheduling.Quartz.EFCore.Sqlite/SqliteQuartzExtensions.cs

@@ -17,7 +17,7 @@ public static class SqliteQuartzExtensions
     /// <summary>
     /// Configures the <see cref="QuartzFeature"/> to use the SQLite job store.
     /// </summary>
-    public static QuartzFeature UseSqlite(this QuartzFeature feature, string connectionString = Constants.DefaultConnectionString, bool useContextPooling = false)
+    public static QuartzFeature UseSqlite(this QuartzFeature feature, string connectionString = Constants.DefaultConnectionString, bool useContextPooling = false, bool useClustering = false)
     {
         if (useContextPooling)
             feature.Services.AddPooledDbContextFactory<SqliteQuartzDbContext>(options => UseSqlite(connectionString, options));
@@ -30,6 +30,9 @@ public static QuartzFeature UseSqlite(this QuartzFeature feature, string connect
             {
                 store.UseNewtonsoftJsonSerializer();
                 store.UseMicrosoftSQLite(connectionString);
+
+                if (useClustering)
+                    store.UseClustering();
             });
         };
 

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

@@ -30,6 +30,65 @@ public QuartzFeature(IModule module) : base(module)
     /// </summary>
     public Action<QuartzHostedServiceOptions>? ConfigureQuartzHostedService { get; set; } = options => options.WaitForJobsToComplete = true;
 
+    private bool _clusteringIdentityConfigured = false;
+    private string? _schedulerId;
+    private string? _schedulerName;
+
+    /// <summary>
+    /// Configures the scheduler instance ID and name for clustered operation.
+    /// </summary>
+    /// <param name="instanceId">The instance ID to use. Use "AUTO" (default) for automatic generation, or specify a unique identifier.</param>
+    /// <param name="schedulerName">The scheduler name. Default is "ElsaScheduler".</param>
+    /// <remarks>
+    /// <para>
+    /// This method configures the scheduler instance ID and name, which are essential for clustered operation.
+    /// When using "AUTO" for the instance ID, Quartz.NET will generate a unique identifier for each scheduler instance.
+    /// </para>
+    /// <para>
+    /// <strong>Note:</strong> This method is optional. If not called, default values will be automatically applied
+    /// (SchedulerId = "AUTO", SchedulerName = "ElsaScheduler"). Only use this method if you need to customize these values.
+    /// </para>
+    /// <para>
+    /// <strong>Important:</strong> This method only configures the scheduler identity settings. To fully enable clustering, you must also:
+    /// 1. Use a persistent job store (e.g., via UseSqlServer, UsePostgreSql, UseMySql, etc.)
+    /// 2. Enable clustering on the persistent store (e.g., useClustering=true parameter)
+    /// </para>
+    /// <para>
+    /// Without both a persistent job store and clustering enabled on that store, clustering cannot function properly.
+    /// The persistent store provides the shared state required for cluster coordination.
+    /// </para>
+    /// <para>
+    /// Example usage with defaults:
+    /// <code>
+    /// .UseQuartz(quartz => quartz.UseSqlServer(connectionString, useClustering: true))
+    /// </code>
+    /// </para>
+    /// <para>
+    /// Example usage with custom identity:
+    /// <code>
+    /// .UseQuartz(quartz => quartz
+    ///     .ConfigureClusteringIdentity("MyCustomId", "MyScheduler")
+    ///     .UseSqlServer(connectionString, useClustering: true))
+    /// </code>
+    /// </para>
+    /// </remarks>
+    public QuartzFeature ConfigureClusteringIdentity(
+        string instanceId = "AUTO",
+        string schedulerName = "ElsaScheduler")
+    {
+        _clusteringIdentityConfigured = true;
+        _schedulerId = instanceId;
+        _schedulerName = schedulerName;
+
+        ConfigureQuartz += quartz =>
+        {
+            quartz.SchedulerId = instanceId;
+            quartz.SchedulerName = schedulerName;
+        };
+
+        return this;
+    }
+
     /// <inheritdoc />
     public override void ConfigureHostedServices()
     {
@@ -46,6 +105,23 @@ public override void Apply()
         if (ConfigureQuartzOptions != null)
             Services.Configure(ConfigureQuartzOptions);
 
+        // Auto-configure clustering identity with default values if not explicitly configured
+        if (!_clusteringIdentityConfigured)
+        {
+            ConfigureQuartz += quartz =>
+            {
+                if (string.IsNullOrEmpty(_schedulerId))
+                    quartz.SchedulerId = "AUTO";
+                else
+                    quartz.SchedulerId = _schedulerId;
+
+                if (string.IsNullOrEmpty(_schedulerName))
+                    quartz.SchedulerName = "ElsaScheduler";
+                else
+                    quartz.SchedulerName = _schedulerName;
+            };
+        }
+
         Services
             .AddQuartz(configure => { ConfigureQuartzInternal(configure, ConfigureQuartz); });
     }

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

@@ -3,6 +3,7 @@
 using Elsa.Extensions;
 using Elsa.Scheduling.Quartz.Contracts;
 using Elsa.Scheduling.Quartz.Jobs;
+using Microsoft.Extensions.Logging;
 using Quartz;
 using QuartzIScheduler = Quartz.IScheduler;
 
@@ -11,7 +12,7 @@ namespace Elsa.Scheduling.Quartz.Services;
 /// <summary>
 /// An implementation of <see cref="IWorkflowScheduler"/> that uses Quartz.NET.
 /// </summary>
-public class QuartzWorkflowScheduler(ISchedulerFactory schedulerFactoryFactory, IJsonSerializer jsonSerializer, ITenantAccessor tenantAccessor, IJobKeyProvider jobKeyProvider) : IWorkflowScheduler
+public class QuartzWorkflowScheduler(ISchedulerFactory schedulerFactoryFactory, IJsonSerializer jsonSerializer, ITenantAccessor tenantAccessor, IJobKeyProvider jobKeyProvider, ILogger<QuartzWorkflowScheduler> logger) : IWorkflowScheduler
 {
     /// <inheritdoc />
     public async ValueTask ScheduleAtAsync(string taskName, ScheduleNewWorkflowInstanceRequest request, DateTimeOffset at, CancellationToken cancellationToken = default)
@@ -109,8 +110,21 @@ public async ValueTask UnscheduleAsync(string taskName, CancellationToken cancel
 
     private async Task ScheduleJobAsync(QuartzIScheduler scheduler, ITrigger trigger, CancellationToken cancellationToken)
     {
-        if (!await scheduler.CheckExists(trigger.Key, cancellationToken))
+        try
+        {
+            // Try to schedule the trigger. In clustered mode, multiple instances may attempt this simultaneously.
+            // The ScheduleJob method will throw ObjectAlreadyExistsException if a trigger with the same key already exists.
+            // Unlike AddJob, ScheduleJob does not have a 'replace' parameter - it always fails if the trigger exists.
+            // Note: To update an existing trigger, callers should first use UnscheduleAsync before scheduling the new trigger.
             await scheduler.ScheduleJob(trigger, cancellationToken);
+        }
+        catch (ObjectAlreadyExistsException)
+        {
+            // Trigger already exists. In clustered scenarios, this is an expected race condition
+            // when multiple pods attempt to schedule the same trigger during tenant activation or startup.
+            // We can safely ignore this and continue, as the trigger is already scheduled.
+            logger.LogDebug("Trigger {TriggerKey} already exists, skipping scheduling. This is expected in clustered deployments during concurrent operations", trigger.Key);
+        }
     }
 
     private JobDataMap CreateJobDataMap(ScheduleNewWorkflowInstanceRequest request)

src/modules/scheduling/Elsa.Scheduling.Quartz/Tasks/RegisterJobsTask.cs

@@ -2,6 +2,7 @@
 using Elsa.Scheduling.Quartz.Contracts;
 using Elsa.Scheduling.Quartz.Jobs;
 using JetBrains.Annotations;
+using Microsoft.Extensions.Logging;
 using Quartz;
 using QuartzIScheduler = Quartz.IScheduler;
 
@@ -12,8 +13,9 @@ namespace Elsa.Scheduling.Quartz.Tasks;
 /// </summary>
 /// <param name="schedulerFactoryFactory"></param>
 /// <param name="jobKeyProvider"></param>
+/// <param name="logger"></param>
 [UsedImplicitly]
-internal class RegisterJobsTask(ISchedulerFactory schedulerFactoryFactory, IJobKeyProvider jobKeyProvider) : IStartupTask
+internal class RegisterJobsTask(ISchedulerFactory schedulerFactoryFactory, IJobKeyProvider jobKeyProvider, ILogger<RegisterJobsTask> logger) : IStartupTask
 {
     public async Task ExecuteAsync(CancellationToken cancellationToken)
     {
@@ -30,7 +32,18 @@ private async Task CreateJobAsync<TJobType>(QuartzIScheduler scheduler, Cancella
             .StoreDurably()
             .Build();
 
-        if (!await scheduler.CheckExists(job.Key, cancellationToken))
-            await scheduler.AddJob(job, false, cancellationToken);
+        try
+        {
+            // Try to add the job. In clustered mode, multiple instances may attempt this simultaneously.
+            // Use replace=false to ensure we don't overwrite an existing job definition.
+            const bool replaceExisting = false;
+            await scheduler.AddJob(job, replaceExisting, cancellationToken);
+        }
+        catch (ObjectAlreadyExistsException)
+        {
+            // Job already exists, which is fine in clustered scenarios where multiple pods
+            // may start concurrently. This is an expected race condition and can be safely ignored.
+            logger.LogDebug("Job {JobKey} already exists, skipping registration. This is expected in clustered deployments during concurrent startup", key);
+        }
     }
 }