Skip to content

Reduce bookmark scheduling backlog startup latency #7730

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sfmskywalker wants to merge 5 commits into
base: release/3.8.0
Choose a base branch
from
+122 −103
Open

Reduce bookmark scheduling backlog startup latency #7730

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d9f2453
Parallelize bookmark scheduling startup backlog
sfmskywalker Jun 16, 2026
a1c446a
Add bookmark scheduler concurrency coverage
sfmskywalker Jun 16, 2026
b7cd6cf
Clean up bookmark scheduler concurrency test
sfmskywalker Jun 16, 2026
01e8e14
Preserve async bookmark scheduler semantics
sfmskywalker Jun 16, 2026
3eb4b9a
Normalize bookmark scheduler file endings
sfmskywalker Jun 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
154 changes: 51 additions & 103 deletions src/modules/Elsa.Scheduling/Services/DefaultBookmarkScheduler.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ namespace Elsa.Scheduling.Services;
/// </summary>
public class DefaultBookmarkScheduler : IBookmarkScheduler
{
private const int MaxConcurrentSchedulingOperations = 16;
private readonly IWorkflowScheduler _workflowScheduler;

/// <summary>
Expand All @@ -28,58 +29,13 @@ public async Task ScheduleAsync(IEnumerable<StoredBookmark> bookmarks, Cancellat
var startAtBookmarks = bookmarkList.Filter(SchedulingStimulusNames.StartAt);
var timerBookmarks = bookmarkList.Filter(SchedulingStimulusNames.Timer);
var cronBookmarks = bookmarkList.Filter(SchedulingStimulusNames.Cron);
var schedules = delayBookmarks
.Select(bookmark => CreateScheduleAt(bookmark.Id, bookmark.WorkflowInstanceId, bookmark.GetPayload<DelayPayload>().ResumeAt))
.Concat(startAtBookmarks.Select(bookmark => CreateScheduleAt(bookmark.Id, bookmark.WorkflowInstanceId, bookmark.GetPayload<StartAtPayload>().ExecuteAt)))
.Concat(timerBookmarks.Select(bookmark => CreateScheduleAt(bookmark.Id, bookmark.WorkflowInstanceId, bookmark.GetPayload<TimerBookmarkPayload>().ResumeAt)))
.Concat(cronBookmarks.Select(bookmark => CreateCronSchedule(bookmark.Id, bookmark.WorkflowInstanceId, bookmark.GetPayload<CronBookmarkPayload>().CronExpression)));

// Schedule each Delay bookmark.
foreach (var bookmark in delayBookmarks)
{
var payload = bookmark.GetPayload<DelayPayload>();
var resumeAt = payload.ResumeAt;
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = bookmark.WorkflowInstanceId,
BookmarkId = bookmark.Id
};
await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, resumeAt, cancellationToken);
}

// Schedule each StartAt bookmark.
foreach (var bookmark in startAtBookmarks)
{
var payload = bookmark.GetPayload<StartAtPayload>();
var executeAt = payload.ExecuteAt;
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = bookmark.WorkflowInstanceId,
BookmarkId = bookmark.Id
};
await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, executeAt, cancellationToken);
}

// Schedule each Timer bookmark.
foreach (var bookmark in timerBookmarks)
{
var payload = bookmark.GetPayload<TimerBookmarkPayload>();
var resumeAt = payload.ResumeAt;
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = bookmark.WorkflowInstanceId,
BookmarkId = bookmark.Id
};
await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, resumeAt, cancellationToken);
}

// Schedule each Cron bookmark.
foreach (var bookmark in cronBookmarks)
{
var payload = bookmark.GetPayload<CronBookmarkPayload>();
var cronExpression = payload.CronExpression;
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = bookmark.WorkflowInstanceId,
BookmarkId = bookmark.Id
};
await _workflowScheduler.ScheduleCronAsync(bookmark.Id, request, cronExpression, cancellationToken);
}
await ScheduleAsync(schedules, cancellationToken);
}

/// <inheritdoc />
Expand All @@ -90,58 +46,13 @@ public async Task ScheduleAsync(string workflowInstanceId, IEnumerable<Bookmark>
var startAtBookmarks = bookmarkList.Filter(SchedulingStimulusNames.StartAt);
var timerBookmarks = bookmarkList.Filter(SchedulingStimulusNames.Timer);
var cronBookmarks = bookmarkList.Filter(SchedulingStimulusNames.Cron);
var schedules = delayBookmarks
.Select(bookmark => CreateScheduleAt(bookmark.Id, workflowInstanceId, bookmark.GetPayload<DelayPayload>().ResumeAt))
.Concat(startAtBookmarks.Select(bookmark => CreateScheduleAt(bookmark.Id, workflowInstanceId, bookmark.GetPayload<StartAtPayload>().ExecuteAt)))
.Concat(timerBookmarks.Select(bookmark => CreateScheduleAt(bookmark.Id, workflowInstanceId, bookmark.GetPayload<TimerBookmarkPayload>().ResumeAt)))
.Concat(cronBookmarks.Select(bookmark => CreateCronSchedule(bookmark.Id, workflowInstanceId, bookmark.GetPayload<CronBookmarkPayload>().CronExpression)));

// Schedule each Delay bookmark.
foreach (var bookmark in delayBookmarks)
{
var payload = bookmark.GetPayload<DelayPayload>();
var resumeAt = payload.ResumeAt;
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = workflowInstanceId,
BookmarkId = bookmark.Id
};
await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, resumeAt, cancellationToken);
}

// Schedule each StartAt bookmark.
foreach (var bookmark in startAtBookmarks)
{
var payload = bookmark.GetPayload<StartAtPayload>();
var executeAt = payload.ExecuteAt;
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = workflowInstanceId,
BookmarkId = bookmark.Id
};
await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, executeAt, cancellationToken);
}

// Schedule each Timer bookmark.
foreach (var bookmark in timerBookmarks)
{
var payload = bookmark.GetPayload<TimerBookmarkPayload>();
var resumeAt = payload.ResumeAt;
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = workflowInstanceId,
BookmarkId = bookmark.Id
};
await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, resumeAt, cancellationToken);
}

// Schedule each Cron bookmark.
foreach (var bookmark in cronBookmarks)
{
var payload = bookmark.GetPayload<CronBookmarkPayload>();
var cronExpression = payload.CronExpression;
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = workflowInstanceId,
BookmarkId = bookmark.Id
};
await _workflowScheduler.ScheduleCronAsync(bookmark.Id, request, cronExpression, cancellationToken);
}
await ScheduleAsync(schedules, cancellationToken);
}

/// <inheritdoc />
Expand Down Expand Up @@ -171,4 +82,41 @@ public async Task UnscheduleAsync(IEnumerable<StoredBookmark> bookmarks, Cancell
foreach (var bookmark in bookmarksToUnSchedule)
await _workflowScheduler.UnscheduleAsync(bookmark.Id, cancellationToken);
}
}

private Task ScheduleAsync(IEnumerable<BookmarkSchedule> schedules, CancellationToken cancellationToken)
{
var options = new ParallelOptions
{
MaxDegreeOfParallelism = MaxConcurrentSchedulingOperations,
CancellationToken = cancellationToken
};

return Parallel.ForEachAsync(schedules, options, ScheduleBookmarkAsync);
}

private async ValueTask ScheduleBookmarkAsync(BookmarkSchedule schedule, CancellationToken cancellationToken)
{
var request = new ScheduleExistingWorkflowInstanceRequest
{
WorkflowInstanceId = schedule.WorkflowInstanceId,
BookmarkId = schedule.BookmarkId
};

if (schedule.CronExpression != null)
await _workflowScheduler.ScheduleCronAsync(schedule.BookmarkId, request, schedule.CronExpression, cancellationToken);
else
await _workflowScheduler.ScheduleAtAsync(schedule.BookmarkId, request, schedule.ExecuteAt, cancellationToken);
}

private static BookmarkSchedule CreateScheduleAt(string bookmarkId, string workflowInstanceId, DateTimeOffset executeAt)
{
return new BookmarkSchedule(bookmarkId, workflowInstanceId, executeAt, null);
}

private static BookmarkSchedule CreateCronSchedule(string bookmarkId, string workflowInstanceId, string cronExpression)
{
return new BookmarkSchedule(bookmarkId, workflowInstanceId, default, cronExpression);
}

private readonly record struct BookmarkSchedule(string BookmarkId, string WorkflowInstanceId, DateTimeOffset ExecuteAt, string? CronExpression);
}
71 changes: 71 additions & 0 deletions test/unit/Elsa.Scheduling.UnitTests/Services/DefaultBookmarkSchedulerTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
using Elsa.Scheduling.Bookmarks;
using Elsa.Scheduling.Services;
using Elsa.Workflows.Runtime.Entities;
using NSubstitute;

namespace Elsa.Scheduling.UnitTests.Services;

public class DefaultBookmarkSchedulerTests
{
private readonly IWorkflowScheduler _workflowScheduler;
private readonly DefaultBookmarkScheduler _scheduler;

public DefaultBookmarkSchedulerTests()
{
_workflowScheduler = Substitute.For<IWorkflowScheduler>();
_scheduler = new DefaultBookmarkScheduler(_workflowScheduler);
}

[Fact]
public async Task ScheduleAsync_WithStoredBookmarks_StartsSchedulingConcurrently()
{
var releaseScheduler = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var scheduledCount = 0;
var bookmarks = CreateStoredDelayBookmarks(2);
_workflowScheduler
.ScheduleAtAsync(Arg.Any<string>(), Arg.Any<ScheduleExistingWorkflowInstanceRequest>(), Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>())
.Returns(_ =>
{
Interlocked.Increment(ref scheduledCount);
return new ValueTask(releaseScheduler.Task);
});

var scheduleTask = _scheduler.ScheduleAsync(bookmarks, CancellationToken.None);

var allSchedulingStarted = await WaitUntilAsync(() => Volatile.Read(ref scheduledCount) == bookmarks.Count, TimeSpan.FromSeconds(5));
var completedBeforeRelease = scheduleTask.IsCompleted;
releaseScheduler.SetResult();
await scheduleTask;

Assert.True(allSchedulingStarted);
Assert.False(completedBeforeRelease);
}

private static IReadOnlyCollection<StoredBookmark> CreateStoredDelayBookmarks(int count)
{
var now = DateTimeOffset.UtcNow;
return Enumerable.Range(1, count).Select(index => new StoredBookmark
{
Id = $"bookmark-{index}",
Name = SchedulingStimulusNames.Delay,
Hash = $"hash-{index}",
WorkflowInstanceId = $"workflow-{index}",
Payload = new DelayPayload(now.AddMinutes(index)),
CreatedAt = now
}).ToList();
}

private static async Task<bool> WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
{
var stopAt = DateTimeOffset.UtcNow.Add(timeout);
while (DateTimeOffset.UtcNow < stopAt)
{
if (condition())
return true;

await Task.Delay(10);
}

return condition();
}
}
Loading