Enhances workflow runtime resilience and signaling (#6643)

* Refactor BookmarkQueueSignaler to use Channel for signaling.

Replaced TaskCompletionSource with a bounded Channel to improve concurrency control and simplify the code. This change ensures better handling of multiple producers while maintaining a single reader model.

* Refactor BookmarkQueueWorker to improve resilience and clarity

Replaced Debouncer with Throttler for rate limiting and added error handling to log exceptions and ensure the worker loop continues safely while allowing proper shutdown on cancellation.

* Handle missing workflow instance during bookmark resumption

Add exception handling for `WorkflowInstanceNotFoundException` to handle cases where the workflow instance does not exist. Log a debug message and return `ResumeBookmarkResult.NotFound()` when such instances are encountered. This ensures better error management and logging for bookmark resumption.

* Refactor default initializations and rename completion methods.

Replaced `default!` with `null!` for input properties to improve clarity and consistency. Renamed methods to better reflect their purpose, changing `CheckIfCompletedAsync` to `AttemptToCompleteAsync`. These changes enhance code readability and maintainability.

* Refactor to use specific exceptions for workflow errors

Replaced generic `InvalidOperationException` with `WorkflowInstanceNotFoundException` and `WorkflowGraphNotFoundException` for improved error context. This enhances clarity and enables more precise error handling.

* Change default value of WorkflowInstanceId to null

Updated the property `WorkflowInstanceId` to use `null!` instead of `default!` to better align with nullable reference type semantics. This ensures clarity and consistency in the codebase regarding expected default values.

* Add handling for WorkflowInstanceSaved in SignalBookmarkQueueWorker

Updated the SignalBookmarkQueueWorker to implement INotificationHandler for WorkflowInstanceSaved. This ensures that workflow instance save events now trigger the bookmark queue worker, improving event handling consistency.

* Update comment to clarify bookmark and workflow instance check

Expanded the comment to explain that the queue item is stored not only when a bookmark is missing but also when the associated workflow instance is not yet in the database. This improves clarity for future maintainers regarding queuing conditions.

src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs

@@ -30,7 +30,7 @@ public class BulkDispatchWorkflows : Activity
     private const string CompletedInstancesCountKey = nameof(CompletedInstancesCountKey);
 
     /// <inheritdoc />
-    public BulkDispatchWorkflows([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
+    public BulkDispatchWorkflows([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
     {
     }
 
@@ -42,13 +42,13 @@ public class BulkDispatchWorkflows : Activity
         Description = "The definition ID of the workflows to dispatch.",
         UIHint = InputUIHints.WorkflowDefinitionPicker
     )]
-    public Input<string> WorkflowDefinitionId { get; set; } = default!;
+    public Input<string> WorkflowDefinitionId { get; set; } = null!;
 
     /// <summary>
     /// The data source to use for dispatching the workflows.
     /// </summary>
     [Input(Description = "The data source to use for dispatching the workflows.")]
-    public Input<object> Items { get; set; } = default!;
+    public Input<object> Items { get; set; } = null!;
 
     /// <summary>
     /// The default key to use for the item input. Will not be used if the Items contain a list of dictionaries.
@@ -69,7 +69,7 @@ public class BulkDispatchWorkflows : Activity
     /// The input to send to the workflows.
     /// </summary>
     [Input(Description = "Additional input to send to the workflows being dispatched.")]
-    public Input<IDictionary<string, object>?> Input { get; set; } = default!;
+    public Input<IDictionary<string, object>?> Input { get; set; } = null!;
 
     /// <summary>
     /// True to wait for the child workflow to complete before completing this activity, false to "fire and forget".
@@ -78,12 +78,12 @@ public class BulkDispatchWorkflows : Activity
         Description = "Wait for the dispatched workflows to complete before completing this activity.",
         DefaultValue = true)]
     public Input<bool> WaitForCompletion { get; set; } = new(true);
-    
+
     /// <summary>
     /// Indicates whether a new trace context should be started for the workflow execution.
     /// </summary>
     [Input(Description = "Start a new trace context when using Open Telemetry.", Category = "Open Telemetry")]
-    public Input<bool> StartNewTrace { get; set; }
+    public Input<bool> StartNewTrace { get; set; } = new(false);
 
     /// <summary>
     /// The channel to dispatch the workflow to.
@@ -94,7 +94,7 @@ public class BulkDispatchWorkflows : Activity
         UIHint = InputUIHints.DropDown,
         UIHandler = typeof(DispatcherChannelOptionsProvider)
     )]
-    public Input<string?> ChannelName { get; set; } = default!;
+    public Input<string?> ChannelName { get; set; } = null!;
 
     /// <summary>
     /// An activity to execute when the child workflow finishes.
@@ -238,17 +238,17 @@ public class BulkDispatchWorkflows : Activity
                 await context.ScheduleActivityAsync(ChildCompleted, options);
                 return;
             default:
-                await CheckIfCompletedAsync(context);
+                await AttemptToCompleteAsync(context);
                 break;
         }
     }
 
     private async ValueTask OnChildFinishedCompletedAsync(ActivityCompletedContext context)
     {
-        await CheckIfCompletedAsync(context.TargetContext);
+        await AttemptToCompleteAsync(context.TargetContext);
     }
 
-    private async ValueTask CheckIfCompletedAsync(ActivityExecutionContext context)
+    private async ValueTask AttemptToCompleteAsync(ActivityExecutionContext context)
     {
         var dispatchedInstancesCount = context.GetProperty<long>(DispatchedInstancesCountKey);
         var finishedInstancesCount = context.GetProperty<long>(CompletedInstancesCountKey);

src/modules/Elsa.Workflows.Runtime/Exceptions/WorkflowInstanceNotFoundException.cs

@@ -0,0 +1,6 @@
+namespace Elsa.Workflows.Runtime.Exceptions;
+
+public class WorkflowInstanceNotFoundException(string message, string instanceId) : Exception(message)
+{
+    public string InstanceId { get; } = instanceId;
+}

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

@@ -1,4 +1,5 @@
 using Elsa.Mediator.Contracts;
+using Elsa.Workflows.Management.Notifications;
 using Elsa.Workflows.Runtime.Notifications;
 using JetBrains.Annotations;
 
@@ -8,7 +9,7 @@ namespace Elsa.Workflows.Runtime.Handlers;
 /// Signals the bookmark queue worker to process any queued work.
 /// </summary>
 [UsedImplicitly]
-public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotificationHandler<WorkflowBookmarksIndexed>, INotificationHandler<BookmarkSaved>
+public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotificationHandler<WorkflowBookmarksIndexed>, INotificationHandler<BookmarkSaved>, INotificationHandler<WorkflowInstanceSaved>
 {
     public Task HandleAsync(BookmarkSaved notification, CancellationToken cancellationToken)
     {
@@ -20,6 +21,11 @@ public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotif
         return Trigger();
     }
 
+    public Task HandleAsync(WorkflowInstanceSaved notification, CancellationToken cancellationToken)
+    {
+        return Trigger();
+    }
+
     private async Task Trigger()
     {
         await signaler.TriggerAsync();

src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs

@@ -10,7 +10,7 @@ public record RunWorkflowInstanceResponse
     /// <summary>
     /// The ID of the workflow instance.
     /// </summary>
-    public string WorkflowInstanceId { get; set; } = default!; 
+    public string WorkflowInstanceId { get; set; } = null!; 
     
     /// <summary>
     /// The status of the workflow instance.

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

@@ -1,43 +1,30 @@
+using System.Threading.Channels;
+
 namespace Elsa.Workflows.Runtime;
 
 public class BookmarkQueueSignaler : IBookmarkQueueSignaler
 {
-    private readonly object _lock = new();
-    private TaskCompletionSource<object?> _tcs = new();
+    private readonly Channel<object?> _channel;
 
-    public async Task AwaitAsync(CancellationToken cancellationToken)
+    public BookmarkQueueSignaler()
     {
-        Task waitTask;
-        lock (_lock)
+        var options = new BoundedChannelOptions(1)
         {
-            // Capture the current TCS and await it
-            waitTask = _tcs.Task;
-        }
+            SingleReader = true,
+            SingleWriter = false,
+            AllowSynchronousContinuations = false
+        };
+        _channel = Channel.CreateBounded<object?>(options);
+    }
 
-        await WaitAndResetAsync(waitTask);
+    public Task AwaitAsync(CancellationToken cancellationToken)
+    {
+        return _channel.Reader.ReadAsync(cancellationToken).AsTask();
     }
 
     public Task TriggerAsync(CancellationToken cancellationToken)
     {
-        lock (_lock)
-        {
-            // If TCS is already in a completed state, no need to set it again.
-            if (!_tcs.Task.IsCompleted)
-            {
-                _tcs.SetResult(null);
-            }
-        }
-        
+        _channel.Writer.TryWrite(null);
         return Task.CompletedTask;
     }
-
-    private async Task WaitAndResetAsync(Task waitTask)
-    {
-        await waitTask;
-        lock (_lock)
-        {
-            // Reset the TCS for the next wait
-            _tcs = new();
-        }
-    }
 }

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

@@ -7,7 +7,7 @@ namespace Elsa.Workflows.Runtime;
 public class BookmarkQueueWorker : IBookmarkQueueWorker
 {
     private readonly RateLimitedFunc<CancellationToken, Task> _rateLimitedProcessAsync;
-    private CancellationTokenSource _cts = default!;
+    private CancellationTokenSource _cts = null!;
     private bool _running;
     private readonly IBookmarkQueueSignaler _signaler;
     private readonly IServiceScopeFactory _scopeFactory;
@@ -18,7 +18,7 @@ public class BookmarkQueueWorker : IBookmarkQueueWorker
         _signaler = signaler;
         _scopeFactory = scopeFactory;
         _logger = logger;
-        _rateLimitedProcessAsync = Debouncer.Debounce<CancellationToken, Task>(ProcessAsync, TimeSpan.FromMilliseconds(500));
+        _rateLimitedProcessAsync = Throttler.Throttle<CancellationToken, Task>(ProcessAsync, TimeSpan.FromMilliseconds(500));
     }
 
     public void Start()
@@ -47,8 +47,19 @@ public class BookmarkQueueWorker : IBookmarkQueueWorker
     {
         while (!_cts.IsCancellationRequested)
         {
-            await _signaler.AwaitAsync(_cts.Token);
-            await _rateLimitedProcessAsync.InvokeAsync(_cts.Token);
+            try
+            {
+                await _signaler.AwaitAsync(_cts.Token);
+                await _rateLimitedProcessAsync.InvokeAsync(_cts.Token);
+            }
+            catch (OperationCanceledException)
+            {
+                break; // Stop() was called
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "BookmarkQueueWorker error – continuing loop");
+            }
         }
     }
 

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

@@ -1,4 +1,5 @@
 using Elsa.Workflows.Helpers;
+using Elsa.Workflows.Runtime.Exceptions;
 using Elsa.Workflows.Runtime.Filters;
 using Elsa.Workflows.Runtime.Messages;
 using Elsa.Workflows.Runtime.Options;
@@ -64,7 +65,7 @@ public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bo
             ActivityHandle = request.ActivityHandle,
             BookmarkId = request.BookmarkId
         };
-        
+
         var workflowInstanceId = request.WorkflowInstanceId;
         var workflowClient = await workflowRuntime.CreateClientAsync(workflowInstanceId, cancellationToken);
         var response = await workflowClient.RunInstanceAsync(runRequest, cancellationToken);
@@ -89,8 +90,18 @@ public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bo
             Properties = options?.Properties,
             BookmarkId = bookmark.Id
         };
-        var response = await workflowClient.RunInstanceAsync(runRequest, cancellationToken);
-        logger.LogDebug("Resumed workflow instance {WorkflowInstanceId} with bookmark {BookmarkId}", bookmark.WorkflowInstanceId, bookmark.Id);
-        return ResumeBookmarkResult.Found(response);
+
+        try
+        {
+            var response = await workflowClient.RunInstanceAsync(runRequest, cancellationToken);
+            logger.LogDebug("Resumed workflow instance {WorkflowInstanceId} with bookmark {BookmarkId}", bookmark.WorkflowInstanceId, bookmark.Id);
+            return ResumeBookmarkResult.Found(response);
+        }
+        catch (WorkflowInstanceNotFoundException)
+        {
+            // The workflow instance does not (yet) exist in the DB.
+            logger.LogDebug("No workflow instance with ID {WorkflowInstanceId} found for bookmark {BookmarkId} at this time.", bookmark.WorkflowInstanceId, bookmark.Id);
+            return ResumeBookmarkResult.NotFound();
+        }
     }
 }

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

@@ -4,6 +4,7 @@ using Elsa.Workflows.Management.Mappers;
 using Elsa.Workflows.Management.Options;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Options;
+using Elsa.Workflows.Runtime.Exceptions;
 using Elsa.Workflows.Runtime.Messages;
 using Elsa.Workflows.State;
 using Microsoft.Extensions.Logging;
@@ -164,7 +165,7 @@ public class LocalWorkflowClient(
     private async Task<WorkflowInstance> GetWorkflowInstanceAsync(CancellationToken cancellationToken)
     {
         var workflowInstance = await workflowInstanceManager.FindByIdAsync(WorkflowInstanceId, cancellationToken);
-        if (workflowInstance == null) throw new InvalidOperationException($"Workflow instance {WorkflowInstanceId} not found.");
+        if (workflowInstance == null) throw new WorkflowInstanceNotFoundException($"Workflow instance not found.", WorkflowInstanceId);
         return workflowInstance;
     }
 
@@ -177,7 +178,7 @@ public class LocalWorkflowClient(
     private async Task<WorkflowGraph> GetWorkflowGraphAsync(WorkflowDefinitionHandle definitionHandle, CancellationToken cancellationToken)
     {
         var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionHandle, cancellationToken);
-        if (workflowGraph == null) throw new InvalidOperationException($"Workflow graph with handle {definitionHandle} not found.");
+        if (workflowGraph == null) throw new WorkflowGraphNotFoundException($"Workflow graph not found.", definitionHandle);
         return workflowGraph;
     }
 }

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

@@ -32,7 +32,7 @@ public class StoreBookmarkQueue(
             return;
         }
 
-        // There was no matching bookmark yet. Store the queue item for the system to pick up whenever the bookmark becomes present.
+        // There was no matching bookmark yet, or the associated workflow instance hasn't been stored in the DB yet. Store the queue item for the system to pick up whenever the bookmark or workflow instance becomes present.
         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