Fix race condition and add rate limiting to bookmark queue processing (#6187)

* Fix race condition and add rate limiting to bookmark queue processing

Introduce rate-limited function invocation for bookmark queue processing using ThrottleDebounce library to optimize performance. Update related classes and interfaces to support asynchronous and cancellation-aware operations, improving system responsiveness. Adjust logging to provide more detailed information during bookmark queue handling.

* Enable workflows on bug branches

This change updates the GitHub Actions workflow configuration to trigger on branches with the 'bug/*' pattern. This allows for automated actions on bug fix branches alongside the main branch, improving development and testing processes.

.github/workflows/packages.yml

@@ -4,6 +4,7 @@ on:
   push:
     branches:
       - 'main'
+      - 'bug/*'
   release:
     types: [ prereleased ]
 env:

src/apps/Elsa.Server.Web/Program.cs

@@ -337,6 +337,7 @@ services
             {
                 options.AllowClrAccess = true;
                 options.DisableWrappers = disableVariableWrappers;
+                options.RegisterType<OrderReceived>();
                 options.ConfigureEngine(engine =>
                 {
                     engine.Execute("function greet(name) { return `Hello ${name}!`; }");
@@ -560,9 +561,9 @@ services.AddActivityStateFilter<HttpRequestAuthenticationHeaderFilter>();
 // Optionally configure recurring tasks using alternative schedules.
 services.Configure<RecurringTaskOptions>(options =>
 {
-    options.Schedule.ConfigureTask<TriggerBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(30));
+    options.Schedule.ConfigureTask<TriggerBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(300));
+    options.Schedule.ConfigureTask<PurgeBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(300));
     options.Schedule.ConfigureTask<UpdateExpiredSecretsRecurringTask>(TimeSpan.FromHours(4));
-    options.Schedule.ConfigureTask<PurgeBookmarkQueueRecurringTask>(TimeSpan.FromSeconds(11));
 });
 
 services.Configure<BookmarkQueuePurgeOptions>(options => options.Ttl = TimeSpan.FromSeconds(10));

src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs

@@ -8,7 +8,7 @@ public class DistributedBookmarkQueueWorker(
     IDistributedLockProvider distributedLockProvider,
     IBookmarkQueueSignaler signaler,
     IServiceScopeFactory scopeFactory,
-    ILogger<DistributedBookmarkQueueWorker> logger) : BookmarkQueueWorker(signaler, scopeFactory)
+    ILogger<DistributedBookmarkQueueWorker> logger) : BookmarkQueueWorker(signaler, scopeFactory, logger)
 {
     protected override async Task ProcessAsync(CancellationToken cancellationToken)
     {

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

@@ -2,6 +2,6 @@ namespace Elsa.Workflows.Runtime;
 
 public interface IBookmarkQueueSignaler
 {
-    Task AwaitAsync();
-    void Trigger();
+    Task AwaitAsync(CancellationToken cancellationToken = default);
+    Task TriggerAsync(CancellationToken cancellationToken = default);
 }

src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj

@@ -12,6 +12,7 @@
         <PackageReference Include="Microsoft.Extensions.DependencyInjection"/>
         <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions"/>
         <PackageReference Include="Open.Linq.AsyncExtensions"/>
+        <PackageReference Include="ThrottleDebounce" />
     </ItemGroup>
 
     <ItemGroup>

src/modules/Elsa.Workflows.Runtime/Extensions/RateLimitedFuncExtensions.cs

@@ -0,0 +1,31 @@
+using ThrottleDebounce;
+
+namespace Elsa.Workflows.Runtime;
+
+/// <summary>
+/// Adds extension methods for <see cref="RateLimitedFunc{TResult}"/> and <see cref="RateLimitedFunc{T, TResult}"/>.
+/// </summary>
+public static class RateLimitedFuncExtensions
+{
+    /// <summary>
+    /// Invokes the specified rate limited function.
+    /// </summary>
+    public static async Task InvokeAsync(this RateLimitedFunc<Task> rateLimitedFunc)
+    {
+        var task = rateLimitedFunc.Invoke();
+
+        if (task != null)
+            await task;
+    }
+    
+    /// <summary>
+    /// Invokes the specified rate limited function.
+    /// </summary>
+    public static async Task InvokeAsync<T>(this RateLimitedFunc<T, Task> rateLimitedFunc, T arg1)
+    {
+        var task = rateLimitedFunc.Invoke(arg1);
+
+        if (task != null)
+            await task;
+    }
+}

src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs

@@ -230,7 +230,7 @@ public class WorkflowRuntimeFeature : FeatureBase
             .AddScoped(WorkflowExecutionLogSink)
             .AddSingleton(BackgroundActivityScheduler)
             .AddSingleton<RandomLongIdentityGenerator>()
-            .AddScoped<IBookmarkQueueSignaler, BookmarkQueueSignaler>()
+            .AddSingleton<IBookmarkQueueSignaler, BookmarkQueueSignaler>()
             .AddScoped<IBookmarkQueueWorker, BookmarkQueueWorker>()
             .AddScoped<IBookmarkManager, DefaultBookmarkManager>()
             .AddScoped<IActivityExecutionManager, DefaultActivityExecutionManager>()

src/modules/Elsa.Workflows.Runtime/Handlers/SignalBookmarkQueueWorker.cs

@@ -20,9 +20,8 @@ public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotif
         return Trigger();
     }
 
-    private Task Trigger()
+    private async Task Trigger()
     {
-        signaler.Trigger();
-        return Task.CompletedTask;
+        await signaler.TriggerAsync();
     }
 }

src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueProcessor.cs

@@ -3,10 +3,11 @@ using Elsa.Common.Models;
 using Elsa.Extensions;
 using Elsa.Workflows.Runtime.Entities;
 using Elsa.Workflows.Runtime.OrderDefinitions;
+using Microsoft.Extensions.Logging;
 
 namespace Elsa.Workflows.Runtime;
 
-public class BookmarkQueueProcessor(IBookmarkQueueStore store, IBookmarkResumer bookmarkResumer) : IBookmarkQueueProcessor
+public class BookmarkQueueProcessor(IBookmarkQueueStore store, IBookmarkResumer bookmarkResumer, ILogger<BookmarkQueueProcessor> logger) : IBookmarkQueueProcessor
 {
     public async Task ProcessAsync(CancellationToken cancellationToken = default)
     {
@@ -37,11 +38,19 @@ public class BookmarkQueueProcessor(IBookmarkQueueStore store, IBookmarkResumer
     {
         var filter = item.CreateBookmarkFilter();
         var options = item.Options;
+        
+        logger.LogDebug("Processing bookmark queue item {BookmarkQueueItemId} for workflow instance {WorkflowInstanceId} for activity type {ActivityType}", item.Id, item.WorkflowInstanceId, item.ActivityTypeName);
+        
         var result = await bookmarkResumer.ResumeAsync(filter, options, cancellationToken);
 
         if (result.Matched)
         {
+            logger.LogDebug("Successfully resumed workflow instance {WorkflowInstance} using bookmark {BookmarkId} for activity type {ActivityType}", item.WorkflowInstanceId, item.BookmarkId, item.ActivityTypeName);
             await store.DeleteAsync(item.Id, cancellationToken);
         }
+        else
+        {
+            logger.LogDebug("No matching bookmark found for bookmark queue item {BookmarkQueueItemId} for workflow instance {WorkflowInstanceId} for activity type {ActivityType}", item.Id, item.WorkflowInstanceId, item.ActivityTypeName);
+        }
     }
 }

src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueSignaler.cs

@@ -2,16 +2,42 @@ namespace Elsa.Workflows.Runtime;
 
 public class BookmarkQueueSignaler : IBookmarkQueueSignaler
 {
-    private TaskCompletionSource? _tsc;
+    private readonly object _lock = new();
+    private TaskCompletionSource<object?> _tcs = new();
 
-    public Task AwaitAsync()
+    public async Task AwaitAsync(CancellationToken cancellationToken)
     {
-        _tsc ??= new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
-        return _tsc.Task.ContinueWith(_ => _tsc = null);
+        Task waitTask;
+        lock (_lock)
+        {
+            // Capture the current TCS and await it
+            waitTask = _tcs.Task;
+        }
+
+        await WaitAndResetAsync(waitTask);
     }
 
-    public void Trigger()
+    public Task TriggerAsync(CancellationToken cancellationToken)
     {
-        _tsc?.TrySetResult();
+        lock (_lock)
+        {
+            // If TCS is already in a completed state, no need to set it again.
+            if (!_tcs.Task.IsCompleted)
+            {
+                _tcs.SetResult(null);
+            }
+        }
+        
+        return Task.CompletedTask;
+    }
+
+    private async Task WaitAndResetAsync(Task waitTask)
+    {
+        await waitTask;
+        lock (_lock)
+        {
+            // Reset the TCS for the next wait
+            _tcs = new TaskCompletionSource<object?>();
+        }
     }
 }

src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs

@@ -1,11 +1,25 @@
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using ThrottleDebounce;
 
 namespace Elsa.Workflows.Runtime;
 
-public class BookmarkQueueWorker(IBookmarkQueueSignaler signaler, IServiceScopeFactory scopeFactory) : IBookmarkQueueWorker
+public class BookmarkQueueWorker : IBookmarkQueueWorker
 {
+    private readonly RateLimitedFunc<CancellationToken, Task> _rateLimitedProcessAsync;
     private CancellationTokenSource _cts = default!;
     private bool _running;
+    private readonly IBookmarkQueueSignaler _signaler;
+    private readonly IServiceScopeFactory _scopeFactory;
+    private readonly ILogger<BookmarkQueueWorker> _logger;
+    
+    public BookmarkQueueWorker(IBookmarkQueueSignaler signaler, IServiceScopeFactory scopeFactory, ILogger<BookmarkQueueWorker> logger)
+    {
+        _signaler = signaler;
+        _scopeFactory = scopeFactory;
+        _logger = logger;
+        _rateLimitedProcessAsync = Debouncer.Debounce<CancellationToken, Task>(ProcessAsync, TimeSpan.FromMilliseconds(500));
+    }
 
     public void Start()
     {
@@ -33,15 +47,17 @@ public class BookmarkQueueWorker(IBookmarkQueueSignaler signaler, IServiceScopeF
     {
         while (!_cts.IsCancellationRequested)
         {
-            await signaler.AwaitAsync();
-            await ProcessAsync(_cts.Token);
+            await _signaler.AwaitAsync(_cts.Token);
+            await _rateLimitedProcessAsync.InvokeAsync(_cts.Token);
         }
     }
 
     protected virtual async Task ProcessAsync(CancellationToken cancellationToken)
     {
-        using var scope = scopeFactory.CreateScope();
+        _logger.LogDebug("Processing bookmark queue...");
+        using var scope = _scopeFactory.CreateScope();
         var processor = scope.ServiceProvider.GetRequiredService<IBookmarkQueueProcessor>();
         await processor.ProcessAsync(cancellationToken);
+        _logger.LogDebug("Processed bookmark queue.");
     }
 }

src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs

@@ -27,12 +27,12 @@ public class StoreBookmarkQueue(
 
         if (result.Matched)
         {
-            logger.LogDebug("Successfully resumed workflow instance {WorkflowInstance} using bookmark {BookmarkId}", item.WorkflowInstanceId, item.BookmarkId);
+            logger.LogDebug("Successfully resumed workflow instance {WorkflowInstance} using bookmark {BookmarkId} for activity type {ActivityType}", item.WorkflowInstanceId, item.BookmarkId, item.ActivityTypeName);
             return;
         }
 
         // There was no matching bookmark yet. Store the queue item for the system to pick up whenever the bookmark becomes present.
-        logger.LogDebug("No bookmark with ID {BookmarkId} found for workflow {WorkflowInstance}. Adding the request to the bookmark queue", item.BookmarkId, item.WorkflowInstanceId);
+        logger.LogDebug("No bookmark with ID {BookmarkId} found for workflow {WorkflowInstance} for activity type {ActivityType}. Adding the request to the bookmark queue", item.BookmarkId, item.WorkflowInstanceId, item.ActivityTypeName);
         
         var entity = new BookmarkQueueItem
         {
@@ -49,6 +49,6 @@ public class StoreBookmarkQueue(
         await store.AddAsync(entity, cancellationToken);
 
         // Trigger the bookmark queue processor.
-        bookmarkQueueSignaler.Trigger();
+        await bookmarkQueueSignaler.TriggerAsync(cancellationToken);
     }
 }

src/modules/Elsa.Workflows.Runtime/Tasks/TriggerBookmarkQueueRecurringTask.cs

@@ -23,7 +23,6 @@ public class TriggerBookmarkQueueRecurringTask(IBookmarkQueueWorker bookmarkQueu
 
     public Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        signaler.Trigger();
-        return Task.CompletedTask;
+        return signaler.TriggerAsync(stoppingToken);
     }
 }