From 2dcd852833bf6de2ad065225096d55cb3b9bb4a8 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 13 May 2025 13:51:47 +0200 Subject: [PATCH] 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. --- .../Activities/BulkDispatchWorkflows.cs | 20 ++++----- .../WorkflowInstanceNotFoundException.cs | 6 +++ .../Handlers/SignalBookmarkQueueWorker.cs | 8 +++- .../Messages/RunWorkflowInstanceResponse.cs | 2 +- .../Services/BookmarkQueueSignaler.cs | 43 +++++++------------ .../Services/BookmarkQueueWorker.cs | 19 ++++++-- .../Services/BookmarkResumer.cs | 19 ++++++-- .../Services/LocalWorkflowClient.cs | 5 ++- .../Services/StoreBookmarkQueue.cs | 2 +- 9 files changed, 73 insertions(+), 51 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Runtime/Exceptions/WorkflowInstanceNotFoundException.cs diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index 2207e05bd..767ee8995 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -30,7 +30,7 @@ public class BulkDispatchWorkflows : Activity private const string CompletedInstancesCountKey = nameof(CompletedInstancesCountKey); /// - 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 WorkflowDefinitionId { get; set; } = default!; + public Input WorkflowDefinitionId { get; set; } = null!; /// /// The data source to use for dispatching the workflows. /// [Input(Description = "The data source to use for dispatching the workflows.")] - public Input Items { get; set; } = default!; + public Input Items { get; set; } = null!; /// /// 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. /// [Input(Description = "Additional input to send to the workflows being dispatched.")] - public Input?> Input { get; set; } = default!; + public Input?> Input { get; set; } = null!; /// /// 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 WaitForCompletion { get; set; } = new(true); - + /// /// Indicates whether a new trace context should be started for the workflow execution. /// [Input(Description = "Start a new trace context when using Open Telemetry.", Category = "Open Telemetry")] - public Input StartNewTrace { get; set; } + public Input StartNewTrace { get; set; } = new(false); /// /// The channel to dispatch the workflow to. @@ -94,7 +94,7 @@ public class BulkDispatchWorkflows : Activity UIHint = InputUIHints.DropDown, UIHandler = typeof(DispatcherChannelOptionsProvider) )] - public Input ChannelName { get; set; } = default!; + public Input ChannelName { get; set; } = null!; /// /// 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(DispatchedInstancesCountKey); var finishedInstancesCount = context.GetProperty(CompletedInstancesCountKey); diff --git a/src/modules/Elsa.Workflows.Runtime/Exceptions/WorkflowInstanceNotFoundException.cs b/src/modules/Elsa.Workflows.Runtime/Exceptions/WorkflowInstanceNotFoundException.cs new file mode 100644 index 000000000..b628a7695 --- /dev/null +++ b/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; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/SignalBookmarkQueueWorker.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/SignalBookmarkQueueWorker.cs index c89ed86ec..10180e72c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Handlers/SignalBookmarkQueueWorker.cs +++ b/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. /// [UsedImplicitly] -public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotificationHandler, INotificationHandler +public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotificationHandler, INotificationHandler, INotificationHandler { 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(); diff --git a/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs b/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs index 8a49e2273..7fa83bfbc 100644 --- a/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs +++ b/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs @@ -10,7 +10,7 @@ public record RunWorkflowInstanceResponse /// /// The ID of the workflow instance. /// - public string WorkflowInstanceId { get; set; } = default!; + public string WorkflowInstanceId { get; set; } = null!; /// /// The status of the workflow instance. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueSignaler.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueSignaler.cs index 3ab6af9cc..ca59b9af5 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueSignaler.cs +++ b/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 _tcs = new(); + private readonly Channel _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(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(); - } - } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs index 1d6fa8736..7035d569e 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs @@ -7,7 +7,7 @@ namespace Elsa.Workflows.Runtime; public class BookmarkQueueWorker : IBookmarkQueueWorker { private readonly RateLimitedFunc _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(ProcessAsync, TimeSpan.FromMilliseconds(500)); + _rateLimitedProcessAsync = Throttler.Throttle(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"); + } } } diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs index 608b7ee9a..d91c37aa0 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs +++ b/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(); + } } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs index 223468d2c..1146d7c88 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs +++ b/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 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 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; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs b/src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs index 31f31e0b6..b3edde3f1 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs +++ b/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