Fix scheduling startup backlog catch-up

src/modules/Elsa.Persistence.EFCore/Modules/Runtime/BookmarkStore.cs

@@ -1,3 +1,5 @@
+using Elsa.Common.Models;
+using Elsa.Extensions;
 using Elsa.Workflows;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Entities;
@@ -36,6 +38,14 @@ public async ValueTask<IEnumerable<StoredBookmark>> FindManyAsync(BookmarkFilter
         return await store.QueryAsync(filter.Apply, OnLoadAsync, filter.TenantAgnostic, cancellationToken);
     }
 
+    /// <inheritdoc />
+    public async ValueTask<Page<StoredBookmark>> FindManyAsync(BookmarkFilter filter, PageArgs pageArgs, CancellationToken cancellationToken = default)
+    {
+        var count = await store.CountAsync(filter.Apply, filter.TenantAgnostic, cancellationToken);
+        var results = (await store.QueryAsync(query => filter.Apply(query).OrderBy(x => x.Id).Paginate(pageArgs), OnLoadAsync, filter.TenantAgnostic, cancellationToken)).ToList();
+        return Page.Of(results, count);
+    }
+
     /// <inheritdoc />
     public async ValueTask<long> DeleteAsync(BookmarkFilter filter, CancellationToken cancellationToken = default)
     {

src/modules/Elsa.Persistence.EFCore/Modules/Runtime/TriggerStore.cs

@@ -8,7 +8,6 @@
 using Elsa.Workflows.Runtime.Filters;
 using Elsa.Workflows.Runtime.OrderDefinitions;
 using JetBrains.Annotations;
-using Open.Linq.AsyncExtensions;
 
 namespace Elsa.Persistence.EFCore.Modules.Runtime;
 
@@ -50,8 +49,8 @@ public ValueTask<Page<StoredTrigger>> FindManyAsync(TriggerFilter filter, PageAr
 
     public async ValueTask<Page<StoredTrigger>> FindManyAsync<TProp>(TriggerFilter filter, PageArgs pageArgs, StoredTriggerOrder<TProp> order, CancellationToken cancellationToken = default)
     {
-        var count = await store.QueryAsync(filter.Apply, OnLoadAsync, cancellationToken).LongCount();
-        var results = await store.QueryAsync(queryable => filter.Apply(queryable).OrderBy(order).Paginate(pageArgs).OrderBy(order), OnLoadAsync, cancellationToken).ToList();
+        var count = await store.CountAsync(filter.Apply, filter.TenantAgnostic, cancellationToken);
+        var results = (await store.QueryAsync(queryable => filter.Apply(queryable).OrderBy(order).Paginate(pageArgs).OrderBy(order), OnLoadAsync, filter.TenantAgnostic, cancellationToken)).ToList();
         return new(results, count);
     }
 

src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs

@@ -6,6 +6,7 @@
 using Elsa.Features.Services;
 using Elsa.Scheduling.Bookmarks;
 using Elsa.Scheduling.Handlers;
+using Elsa.Scheduling.Options;
 using Elsa.Scheduling.Services;
 using Elsa.Scheduling.StartupTasks;
 using Elsa.Scheduling.TriggerPayloadValidators;
@@ -44,6 +45,7 @@ public override void Apply()
             .AddSingleton<UpdateTenantSchedules>()
             .AddSingleton<ITenantDeletedEvent>(sp => sp.GetRequiredService<UpdateTenantSchedules>())
             .AddSingleton<IScheduler, LocalScheduler>()
+            .AddSingleton<PastDueScheduleStaggerer>()
             .AddSingleton<CronosCronParser>()
             .AddSingleton(CronParser)
             .AddScoped<ITriggerScheduler, DefaultTriggerScheduler>()
@@ -59,6 +61,7 @@ public override void Apply()
             // Graceful shutdown: register scheduled-trigger ingress for diagnostic visibility (FR-006).
             .AddSingleton<Elsa.Workflows.Runtime.IIngressSource, Elsa.Scheduling.IngressSources.ScheduledTriggerIngressSource>();
 
+        Services.Configure<SchedulingOptions>(_ => { });
         Services.Configure<SerializationTypeOptions>(options =>
         {
             options.RegisterTypeAlias(typeof(CronBookmarkPayload), nameof(CronBookmarkPayload));

src/modules/Elsa.Scheduling/Options/SchedulingOptions.cs

@@ -0,0 +1,27 @@
+namespace Elsa.Scheduling.Options;
+
+/// <summary>
+/// Configures local scheduling behavior.
+/// </summary>
+public class SchedulingOptions
+{
+    /// <summary>
+    /// The number of stored triggers and bookmarks to load per batch when rebuilding local schedules on startup.
+    /// </summary>
+    public int StartupSchedulePageSize { get; set; } = 1000;
+
+    /// <summary>
+    /// The minimum delay used when a specific-instant schedule is already due.
+    /// </summary>
+    public TimeSpan MinimumPastDueScheduleDelay { get; set; } = TimeSpan.FromMilliseconds(1);
+
+    /// <summary>
+    /// The spacing between past-due schedules during catch-up.
+    /// </summary>
+    public TimeSpan PastDueScheduleStaggerInterval { get; set; } = TimeSpan.FromMilliseconds(50);
+
+    /// <summary>
+    /// The bounded window over which past-due schedules are distributed.
+    /// </summary>
+    public TimeSpan PastDueScheduleStaggerWindow { get; set; } = TimeSpan.FromMinutes(5);
+}

src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs

@@ -1,9 +1,12 @@
 using Elsa.Common;
 using Elsa.Mediator.Contracts;
 using Elsa.Scheduling.Commands;
+using Elsa.Scheduling.Options;
+using Elsa.Scheduling.Services;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Timer = System.Timers.Timer;
+using OptionsFactory = Microsoft.Extensions.Options.Options;
 
 namespace Elsa.Scheduling.ScheduledTasks;
 
@@ -12,10 +15,12 @@ namespace Elsa.Scheduling.ScheduledTasks;
 /// </summary>
 public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable
 {
+    private static readonly PastDueScheduleStaggerer DefaultPastDueScheduleStaggerer = new(OptionsFactory.Create(new SchedulingOptions()));
     private readonly ITask _task;
     private readonly ISystemClock _systemClock;
     private readonly IServiceScopeFactory _scopeFactory;
     private readonly ILogger<ScheduledSpecificInstantTask> _logger;
+    private readonly PastDueScheduleStaggerer _pastDueScheduleStaggerer;
     private readonly DateTimeOffset _startAt;
     private readonly CancellationTokenSource _cancellationTokenSource;
     private readonly SemaphoreSlim _executionSemaphore = new(1, 1);
@@ -27,12 +32,33 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable
     /// <summary>
     /// Initializes a new instance of <see cref="ScheduledSpecificInstantTask"/>.
     /// </summary>
-    public ScheduledSpecificInstantTask(ITask task, DateTimeOffset startAt, ISystemClock systemClock, IServiceScopeFactory scopeFactory, ILogger<ScheduledSpecificInstantTask> logger)
+    public ScheduledSpecificInstantTask(
+        ITask task,
+        DateTimeOffset startAt,
+        ISystemClock systemClock,
+        IServiceScopeFactory scopeFactory,
+        ILogger<ScheduledSpecificInstantTask> logger)
+        : this(task, startAt, systemClock, scopeFactory, logger, DefaultPastDueScheduleStaggerer)
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of <see cref="ScheduledSpecificInstantTask"/>.
+    /// </summary>
+    [ActivatorUtilitiesConstructor]
+    public ScheduledSpecificInstantTask(
+        ITask task,
+        DateTimeOffset startAt,
+        ISystemClock systemClock,
+        IServiceScopeFactory scopeFactory,
+        ILogger<ScheduledSpecificInstantTask> logger,
+        PastDueScheduleStaggerer pastDueScheduleStaggerer)
     {
         _task = task;
         _systemClock = systemClock;
         _scopeFactory = scopeFactory;
         _logger = logger;
+        _pastDueScheduleStaggerer = pastDueScheduleStaggerer;
         _startAt = startAt;
         _cancellationTokenSource = new();
 
@@ -57,16 +83,14 @@ private void Schedule()
     {
         var now = _systemClock.UtcNow;
         var delay = _startAt - now;
+        var adjustedDelay = _pastDueScheduleStaggerer.GetDelay(delay);
 
-        // Handle edge cases where delay is zero or negative (e.g., due to clock drift, fast execution, or time alignment)
-        // Instead of silently returning, use a minimum delay to ensure the timer fires and workflow continues scheduling
         if (delay <= TimeSpan.Zero)
         {
-            _logger.LogWarning("Calculated delay is {Delay} which is not positive. Using minimum delay of 1ms to ensure timer fires", delay);
-            delay = TimeSpan.FromMilliseconds(1);
+            _logger.LogDebug("Calculated delay is {Delay} which is not positive. Using catch-up delay of {CatchUpDelay}", delay, adjustedDelay);
         }
 
-        _timer = new(delay.TotalMilliseconds)
+        _timer = new(adjustedDelay.TotalMilliseconds)
         {
             Enabled = true
         };

src/modules/Elsa.Scheduling/Services/DefaultTriggerScheduler.cs

@@ -41,14 +41,10 @@ public async Task ScheduleAsync(IEnumerable<StoredTrigger> triggers, Cancellatio
         foreach (var trigger in startAtTriggers)
         {
             var executeAt = trigger.GetPayload<StartAtPayload>().ExecuteAt;
-
-            // If the trigger is in the past, log info and skip scheduling.
+
             if (executeAt < now)
-            {
-                logger.LogInformation("StartAt trigger is in the past. TriggerId: {TriggerId}. ExecuteAt: {ExecuteAt}. Skipping scheduling", trigger.Id, executeAt);
-                continue;
-            }
-
+                logger.LogInformation("StartAt trigger is in the past. TriggerId: {TriggerId}. ExecuteAt: {ExecuteAt}. Scheduling catch-up", trigger.Id, executeAt);
+
             var input = new { ExecuteAt = executeAt }.ToDictionary();
             var request = new ScheduleNewWorkflowInstanceRequest
             {

src/modules/Elsa.Scheduling/Services/PastDueScheduleStaggerer.cs

@@ -0,0 +1,39 @@
+using Elsa.Scheduling.Options;
+using Microsoft.Extensions.Options;
+
+namespace Elsa.Scheduling.Services;
+
+/// <summary>
+/// Distributes already-due schedules over a bounded window to avoid dispatch storms during startup catch-up.
+/// </summary>
+public class PastDueScheduleStaggerer(IOptions<SchedulingOptions> options)
+{
+    private long _sequence;
+
+    public TimeSpan GetDelay(TimeSpan calculatedDelay)
+    {
+        if (calculatedDelay > TimeSpan.Zero)
+            return calculatedDelay;
+
+        var currentOptions = options.Value;
+        var minimumDelay = GetPositiveOrDefault(currentOptions.MinimumPastDueScheduleDelay, TimeSpan.FromMilliseconds(1));
+        var staggerInterval = currentOptions.PastDueScheduleStaggerInterval;
+        var staggerWindow = currentOptions.PastDueScheduleStaggerWindow;
+
+        if (staggerInterval <= TimeSpan.Zero || staggerWindow <= TimeSpan.Zero)
+            return minimumDelay;
+
+        var availableWindow = staggerWindow - minimumDelay;
+
+        if (availableWindow <= TimeSpan.Zero)
+            return minimumDelay;
+
+        var slotCount = Math.Max(1, availableWindow.Ticks / staggerInterval.Ticks + 1);
+        var sequence = Interlocked.Increment(ref _sequence) - 1;
+        var slot = (sequence & long.MaxValue) % slotCount;
+
+        return minimumDelay + TimeSpan.FromTicks(staggerInterval.Ticks * slot);
+    }
+
+    private static TimeSpan GetPositiveOrDefault(TimeSpan value, TimeSpan defaultValue) => value > TimeSpan.Zero ? value : defaultValue;
+}

src/modules/Elsa.Scheduling/ShellFeatures/SchedulingFeature.cs

@@ -5,6 +5,7 @@
 using Elsa.Workflows.Options;
 using Elsa.Scheduling.Bookmarks;
 using Elsa.Scheduling.Handlers;
+using Elsa.Scheduling.Options;
 using Elsa.Scheduling.Services;
 using Elsa.Scheduling.StartupTasks;
 using Elsa.Scheduling.TriggerPayloadValidators;
@@ -41,6 +42,7 @@ public void ConfigureServices(IServiceCollection services)
             .AddSingleton<UpdateTenantSchedules>()
             .AddSingleton<ITenantDeletedEvent>(sp => sp.GetRequiredService<UpdateTenantSchedules>())
             .AddSingleton<IScheduler, LocalScheduler>()
+            .AddSingleton<PastDueScheduleStaggerer>()
             .AddSingleton<CronosCronParser>()
             .AddSingleton(CronParser)
             .AddScoped<ITriggerScheduler, DefaultTriggerScheduler>()
@@ -52,6 +54,7 @@ public void ConfigureServices(IServiceCollection services)
             .AddTriggerPayloadValidator<CronTriggerPayloadValidator, CronTriggerPayload>()
             .AddActivitiesFrom<SchedulingFeature>();
 
+        services.Configure<SchedulingOptions>(_ => { });
         services.Configure<SerializationTypeOptions>(options =>
         {
             options.AddTypeAlias<CronBookmarkPayload>();

src/modules/Elsa.Scheduling/StartupTasks/CreateSchedulesStartupTask.cs

@@ -1,17 +1,20 @@
 using Elsa.Common;
 using Elsa.Common.Multitenancy;
+using Elsa.Common.Models;
+using Elsa.Scheduling.Options;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Filters;
 using Elsa.Workflows.Runtime.Tasks;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
 
 namespace Elsa.Scheduling.StartupTasks;
 
 /// <summary>
 /// Enqueues schedule creation when using the default scheduler, which doesn't have its own persistence layer like Quartz or Hangfire.
 /// </summary>
 [TaskDependency(typeof(PopulateRegistriesStartupTask))]
-public class CreateSchedulesStartupTask(IServiceProvider serviceProvider) : IStartupTask
+public class CreateSchedulesStartupTask(IServiceProvider serviceProvider, IOptions<SchedulingOptions> options) : IStartupTask
 {
     public async Task ExecuteAsync(CancellationToken cancellationToken)
     {
@@ -23,12 +26,13 @@ public async Task ExecuteAsync(CancellationToken cancellationToken)
             await CreateSchedulesAsync(serviceProvider, cancellationToken);
     }
 
-    private static async Task CreateSchedulesAsync(IServiceProvider serviceProvider, CancellationToken cancellationToken)
+    private async Task CreateSchedulesAsync(IServiceProvider serviceProvider, CancellationToken cancellationToken)
     {
         var triggerStore = serviceProvider.GetRequiredService<ITriggerStore>();
         var bookmarkStore = serviceProvider.GetRequiredService<IBookmarkStore>();
         var triggerScheduler = serviceProvider.GetRequiredService<ITriggerScheduler>();
         var bookmarkScheduler = serviceProvider.GetRequiredService<IBookmarkScheduler>();
+        var pageSize = Math.Max(1, options.Value.StartupSchedulePageSize);
         var stimulusNames = new[]
         {
             SchedulingStimulusNames.Cron, SchedulingStimulusNames.Timer, SchedulingStimulusNames.StartAt, SchedulingStimulusNames.Delay,
@@ -41,10 +45,50 @@ private static async Task CreateSchedulesAsync(IServiceProvider serviceProvider,
         {
             Names = stimulusNames
         };
-        var triggers = (await triggerStore.FindManyAsync(triggerFilter, cancellationToken)).ToList();
-        var bookmarks = (await bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken)).ToList();
 
-        await triggerScheduler.ScheduleAsync(triggers, cancellationToken);
-        await bookmarkScheduler.ScheduleAsync(bookmarks, cancellationToken);
+        await ScheduleTriggersAsync(triggerStore, triggerScheduler, triggerFilter, pageSize, cancellationToken);
+        await ScheduleBookmarksAsync(bookmarkStore, bookmarkScheduler, bookmarkFilter, pageSize, cancellationToken);
+    }
+
+    private static async Task ScheduleTriggersAsync(ITriggerStore triggerStore, ITriggerScheduler triggerScheduler, TriggerFilter triggerFilter, int pageSize, CancellationToken cancellationToken)
+    {
+        var pageArgs = PageArgs.FromRange(0, pageSize);
+
+        while (true)
+        {
+            var page = await triggerStore.FindManyAsync(triggerFilter, pageArgs, cancellationToken);
+
+            if (page.Items.Count == 0)
+                break;
+
+            await triggerScheduler.ScheduleAsync(page.Items, cancellationToken);
+
+            var nextOffset = pageArgs.Offset.GetValueOrDefault() + page.Items.Count;
+            if (nextOffset >= page.TotalCount)
+                break;
+
+            pageArgs = pageArgs.Next();
+        }
+    }
+
+    private static async Task ScheduleBookmarksAsync(IBookmarkStore bookmarkStore, IBookmarkScheduler bookmarkScheduler, BookmarkFilter bookmarkFilter, int pageSize, CancellationToken cancellationToken)
+    {
+        var pageArgs = PageArgs.FromRange(0, pageSize);
+
+        while (true)
+        {
+            var page = await bookmarkStore.FindManyAsync(bookmarkFilter, pageArgs, cancellationToken);
+
+            if (page.Items.Count == 0)
+                break;
+
+            await bookmarkScheduler.ScheduleAsync(page.Items, cancellationToken);
+
+            var nextOffset = pageArgs.Offset.GetValueOrDefault() + page.Items.Count;
+            if (nextOffset >= page.TotalCount)
+                break;
+
+            pageArgs = pageArgs.Next();
+        }
     }
 }

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

@@ -1,3 +1,4 @@
+using Elsa.Common.Models;
 using Elsa.Workflows.Runtime.Entities;
 using Elsa.Workflows.Runtime.Filters;
 
@@ -34,6 +35,14 @@ public interface IBookmarkStore
     /// </summary>
     ValueTask<IEnumerable<StoredBookmark>> FindManyAsync(BookmarkFilter filter, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Returns a page of bookmarks matching the specified filter.
+    /// </summary>
+    /// <remarks>
+    /// Startup backlog catch-up depends on store-backed paging. Implementations should page at the persistence layer instead of materializing all matches in memory.
+    /// </remarks>
+    ValueTask<Page<StoredBookmark>> FindManyAsync(BookmarkFilter filter, PageArgs pageArgs, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Deletes a set of bookmarks matching the specified filter.
     /// </summary>

src/modules/Elsa.Workflows.Runtime/Stores/CachingTriggerStore.cs

@@ -48,13 +48,13 @@ public async ValueTask<IEnumerable<StoredTrigger>> FindManyAsync(TriggerFilter f
 
     public async ValueTask<Page<StoredTrigger>> FindManyAsync(TriggerFilter filter, PageArgs pageArgs, CancellationToken cancellationToken = default)
     {
-        var cacheKey = hasher.Hash(filter);
+        var cacheKey = hasher.Hash(filter, pageArgs);
         return (await GetOrCreateAsync(cacheKey, async () => await decoratedStore.FindManyAsync(filter, pageArgs, cancellationToken)))!;
     }
 
     public async ValueTask<Page<StoredTrigger>> FindManyAsync<TProp>(TriggerFilter filter, PageArgs pageArgs, StoredTriggerOrder<TProp> order, CancellationToken cancellationToken = default)
     {
-        var cacheKey = hasher.Hash(filter);
+        var cacheKey = hasher.Hash(filter, pageArgs, order);
         return (await GetOrCreateAsync(cacheKey, async () => await decoratedStore.FindManyAsync(filter, pageArgs, order, cancellationToken)))!;
     }