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