From c409414152e478d7580c30071c64edbb2a885ca6 Mon Sep 17 00:00:00 2001 From: raymonddenhaan <155616759+raymonddenhaan@users.noreply.github.com> Date: Tue, 23 Jan 2024 12:05:03 +0100 Subject: [PATCH] Workflow cancellation (#4813) * Removed duplicate entries * Prevented workflows and activities from starting when the parent workflow is being cancelled * Added cancellation to execution contexts * Added store for workflow execution contexts * Added cancellation to workflowRuntime * Removed calling BookmarkPersistedHandler when persisting bookmarks. * Added endpoint for bulk cancelling tasks * Added tests for cancelling workflows * Prevented cancelling the cancellation process since it could have unwanted effects --------- Co-authored-by: Sipke Schoorstra --- Elsa.sln | 2 +- .../ScheduleActivityHandler.cs | 2 +- .../Grains/WorkflowInstance.cs | 60 ++++++++- .../Proto/WorkflowInstance.proto | 1 + .../Services/ProtoActorWorkflowRuntime.cs | 10 +- .../BulkDelete/Endpoint.cs | 2 +- .../WorkflowInstances/BulkCancel/Endpoint.cs | 21 +-- .../ActivityExecutionContext.Cancel.cs | 44 +++++++ ...ivityExecutionContext.ExecutionLogEntry.cs | 40 ++++++ .../Contexts/ActivityExecutionContext.cs | 22 +++- .../WorkflowExecutionContext.Cancel.cs | 37 ++++++ ...kflowExecutionContext.ExecutionLogEntry.cs | 36 +++++ .../Contexts/WorkflowExecutionContext.cs | 31 ++++- .../ActivityExecutionContextExtensions.cs | 44 +------ .../WorkflowExecutionContextExtensions.cs | 32 ----- .../DefaultActivityInvokerMiddleware.cs | 14 +- .../Activities/ExceptionHandlingMiddleware.cs | 2 +- ... => DefaultActivitySchedulerMiddleware.cs} | 4 + .../Options/RunWorkflowOptions.cs | 1 + .../Services/WorkflowRunner.cs | 4 + .../Services/WorkflowStateExtractor.cs | 2 +- .../Activities/BulkDispatchWorkflows.cs | 4 + .../Contracts/IBookmarkPersister.cs | 5 +- .../IWorkflowExecutionContextStore.cs | 24 ++++ .../Contracts/IWorkflowRuntime.cs | 7 + .../Features/WorkflowRuntimeFeature.cs | 10 +- ...lowExecutionContextNotificationsHandler.cs | 36 +++++ .../Workflows/PersistBookmarkMiddleware.cs | 4 +- .../Notifications/WorkflowBookmarksIndexed.cs | 2 +- .../WorkflowBookmarksPersisted.cs | 2 +- .../Options/StartWorkflowHostOptions.cs | 5 + .../Services/BookmarkPersister.cs | 24 +--- .../Services/DefaultWorkflowRuntime.cs | 76 ++++++++++- .../MemoryWorkflowExecutionContextStore.cs | 42 ++++++ .../Services/WorkflowHost.cs | 1 + .../Elsa.IntegrationTests.csproj | 2 + .../DefaultRuntimeTests.cs | 102 ++++++++++++++ .../WorkflowCancellation/ProtoActorTests.cs | 124 ++++++++++++++++++ .../Workflows/BulkSuspendedWorkflow.cs | 33 +++++ .../Workflows/ResumeDispatchWorkflow.cs | 21 +++ .../Workflows/SimpleChildWorkflow.cs | 19 +++ .../Workflows/SimpleSuspendedWorkflow.cs | 22 ++++ 42 files changed, 854 insertions(+), 122 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs create mode 100644 src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.ExecutionLogEntry.cs create mode 100644 src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.Cancel.cs create mode 100644 src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.ExecutionLogEntry.cs rename src/modules/Elsa.Workflows.Core/Middleware/Workflows/{DefaultWorkSchedulerMiddleware.cs => DefaultActivitySchedulerMiddleware.cs} (91%) create mode 100644 src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionContextStore.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Handlers/WorkflowExecutionContextNotificationsHandler.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Services/MemoryWorkflowExecutionContextStore.cs create mode 100644 test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/DefaultRuntimeTests.cs create mode 100644 test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/ProtoActorTests.cs create mode 100644 test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/BulkSuspendedWorkflow.cs create mode 100644 test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/ResumeDispatchWorkflow.cs create mode 100644 test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleChildWorkflow.cs create mode 100644 test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleSuspendedWorkflow.cs diff --git a/Elsa.sln b/Elsa.sln index 501605779..cee039477 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -1,4 +1,4 @@ - + Microsoft Visual Studio Solution File, Format Version 12.00 # Visual Studio Version 17 VisualStudioVersion = 17.7.34003.232 diff --git a/src/modules/Elsa.Alterations/AlterationHandlers/ScheduleActivityHandler.cs b/src/modules/Elsa.Alterations/AlterationHandlers/ScheduleActivityHandler.cs index 43a05acb1..727267988 100644 --- a/src/modules/Elsa.Alterations/AlterationHandlers/ScheduleActivityHandler.cs +++ b/src/modules/Elsa.Alterations/AlterationHandlers/ScheduleActivityHandler.cs @@ -28,7 +28,7 @@ public class ScheduleActivityHandler : AlterationHandlerBase { // If the activity is in a faulted state, reset it to Running. if (existingActivityExecutionContext.Status == ActivityStatus.Faulted) - existingActivityExecutionContext.Status = ActivityStatus.Running; + existingActivityExecutionContext.TransitionTo(ActivityStatus.Running); // Schedule the activity execution context. var parentContext = existingActivityExecutionContext.ParentActivityExecutionContext; diff --git a/src/modules/Elsa.ProtoActor/Grains/WorkflowInstance.cs b/src/modules/Elsa.ProtoActor/Grains/WorkflowInstance.cs index 0dd27ad33..21f680608 100644 --- a/src/modules/Elsa.ProtoActor/Grains/WorkflowInstance.cs +++ b/src/modules/Elsa.ProtoActor/Grains/WorkflowInstance.cs @@ -3,16 +3,22 @@ using Elsa.ProtoActor.Extensions; using Elsa.ProtoActor.Mappers; using Elsa.ProtoActor.ProtoBuf; using Elsa.ProtoActor.Snapshots; +using Elsa.Workflows; using Elsa.Workflows.Contracts; +using Elsa.Workflows.Helpers; using Elsa.Workflows.Management.Contracts; using Elsa.Workflows.Management.Mappers; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Requests; using Elsa.Workflows.State; using Microsoft.Extensions.DependencyInjection; using Proto; using Proto.Cluster; using Proto.Persistence; +using CancellationTokens = Elsa.Workflows.Models.CancellationTokens; +using WorkflowStatus = Elsa.Workflows.WorkflowStatus; +using WorkflowSubStatus = Elsa.Workflows.WorkflowSubStatus; namespace Elsa.ProtoActor.Grains; @@ -38,6 +44,8 @@ internal class WorkflowInstance : WorkflowInstanceBase private IWorkflowHost _workflowHost = default!; private WorkflowState _workflowState = default!; + private readonly ICollection _cancellationTokenSources = new List(); + /// public WorkflowInstance( IServiceScopeFactory scopeFactory, @@ -139,6 +147,10 @@ internal class WorkflowInstance : WorkflowInstanceBase var versionOptions = VersionOptions.FromString(request.VersionOptions); var cancellationToken = Context.CancellationToken; + var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + _cancellationTokenSources.Add(cancellationTokenSource); + cancellationToken = cancellationTokenSource.Token; + // Only need to reconstruct a workflow host if not already done so during CanStart. if (_workflowHost == null!) { @@ -155,7 +167,9 @@ internal class WorkflowInstance : WorkflowInstanceBase CorrelationId = correlationId, Input = input, Properties = properties, - TriggerActivityId = request.TriggerActivityId + TriggerActivityId = request.TriggerActivityId, + StatusUpdatedCallback = StatusUpdated, + CancellationTokens = new CancellationTokens(cancellationToken) }; var task = _workflowHost.StartWorkflowAsync(startWorkflowOptions, cancellationToken); @@ -187,6 +201,31 @@ internal class WorkflowInstance : WorkflowInstanceBase }); } + private void StatusUpdated(WorkflowExecutionContext context) + { + _ = Task.Run(async () => await Update(context)); + } + + private async Task Update(WorkflowExecutionContext context) + { + using var scope = _scopeFactory.CreateScope(); + var extractor = scope.ServiceProvider.GetRequiredService(); + var bookmarkPersistor = scope.ServiceProvider.GetRequiredService(); + var workflowState = extractor.Extract(context); + var originalBookmarks = _workflowHost.WorkflowState.Bookmarks; + + _workflowState = workflowState; + + await SaveSnapshotAsync(); + SaveWorkflowInstance(workflowState); + var newBookmarks = workflowState.Bookmarks; + + var diff = Diff.For(originalBookmarks, newBookmarks); + + var bookmarkRequest = new UpdateBookmarksRequest(workflowState.DefinitionId, diff, workflowState.CorrelationId); + await bookmarkPersistor.PersistBookmarksAsync(bookmarkRequest); + } + /// public override Task Stop() { @@ -208,6 +247,10 @@ internal class WorkflowInstance : WorkflowInstanceBase var activityInstanceId = request.ActivityInstanceId.NullIfEmpty(); var activityHash = request.ActivityHash.NullIfEmpty(); var cancellationToken = Context.CancellationToken; + + var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + _cancellationTokenSources.Add(cancellationTokenSource); + cancellationToken = cancellationTokenSource.Token; var resumeWorkflowHostOptions = new ResumeWorkflowHostOptions { @@ -218,7 +261,8 @@ internal class WorkflowInstance : WorkflowInstanceBase ActivityInstanceId = activityInstanceId, ActivityHash = activityHash, Input = _input, - Properties = _properties + Properties = _properties, + CancellationTokens = cancellationToken }; var definitionId = _definitionId; @@ -260,6 +304,18 @@ internal class WorkflowInstance : WorkflowInstanceBase /// public override Task Resume(ResumeWorkflowRequest request) => Task.FromResult(new WorkflowExecutionResponse()); + public override async Task Cancel() + { + if (_workflowState.Status != WorkflowStatus.Finished) + { + _workflowState.SubStatus = WorkflowSubStatus.Cancelled; + _workflowState.Status = WorkflowStatus.Finished; + } + + foreach(var source in _cancellationTokenSources) + source.Cancel(); + } + /// public override async Task ExportState(ExportWorkflowStateRequest request) { diff --git a/src/modules/Elsa.ProtoActor/Proto/WorkflowInstance.proto b/src/modules/Elsa.ProtoActor/Proto/WorkflowInstance.proto index 1e5b533ac..c6570ac8d 100644 --- a/src/modules/Elsa.ProtoActor/Proto/WorkflowInstance.proto +++ b/src/modules/Elsa.ProtoActor/Proto/WorkflowInstance.proto @@ -12,6 +12,7 @@ service WorkflowInstance { rpc Start (StartWorkflowRequest) returns (WorkflowExecutionResponse); rpc Stop (Empty) returns (Empty); rpc Resume (ResumeWorkflowRequest) returns (WorkflowExecutionResponse); + rpc Cancel (Empty) returns (Empty); rpc ExportState(ExportWorkflowStateRequest) returns (ExportWorkflowStateResponse); rpc ImportState(ImportWorkflowStateRequest) returns (ImportWorkflowStateResponse); } \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index c10b9f788..0d47a9625 100644 --- a/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -16,6 +16,7 @@ using Elsa.Workflows.State; using Proto.Cluster; using Bookmark = Elsa.Workflows.Models.Bookmark; using CountRunningWorkflowsRequest = Elsa.Workflows.Runtime.Requests.CountRunningWorkflowsRequest; +using WorkflowStatus = Elsa.Workflows.WorkflowStatus; namespace Elsa.ProtoActor.Services; @@ -246,6 +247,13 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime return result!; } + /// + public async Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken) + { + var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId); + await client.Cancel(cancellationToken); + } + /// public async Task> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) { @@ -296,7 +304,7 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime DefinitionId = request.DefinitionId, Version = request.Version, CorrelationId = request.CorrelationId, - WorkflowStatus = Workflows.WorkflowStatus.Running + WorkflowStatus = WorkflowStatus.Running }; return await _workflowInstanceStore.CountAsync(filter, cancellationToken); } diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDelete/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDelete/Endpoint.cs index 60edf36b7..c7756c25e 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDelete/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDelete/Endpoint.cs @@ -22,7 +22,7 @@ internal class BulkDelete : ElsaEndpoint public override async Task ExecuteAsync(Request request, CancellationToken cancellationToken) { - var count = await _workflowDefinitionManager.BulkDeleteByDefinitionIdsAsync(request!.DefinitionIds, cancellationToken); + var count = await _workflowDefinitionManager.BulkDeleteByDefinitionIdsAsync(request.DefinitionIds, cancellationToken); return new Response(count); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/BulkCancel/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/BulkCancel/Endpoint.cs index 19a2bb69d..1f25f003d 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/BulkCancel/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/BulkCancel/Endpoint.cs @@ -1,25 +1,30 @@ -using System.Text.Json.Serialization; using Elsa.Abstractions; +using Elsa.Workflows.Runtime.Contracts; namespace Elsa.Workflows.Api.Endpoints.WorkflowInstances.BulkCancel; public class BulkCancel : ElsaEndpoint { + private readonly IWorkflowRuntime _workflowRuntime; + + public BulkCancel(IWorkflowRuntime workflowRuntime) + { + _workflowRuntime = workflowRuntime; + } + public override void Configure() { Post("/bulk-actions/cancel/workflow-instances/by-id"); ConfigurePermissions("cancel:workflow-instances"); } - + public override async Task ExecuteAsync(Request request, CancellationToken cancellationToken) { - // TODO: Implement workflow cancellation. - var count = -1; + var tasks = request.Ids.Select(id => _workflowRuntime.CancelWorkflowAsync(id, cancellationToken)).ToList(); + await Task.WhenAll(tasks); + + var count = tasks.Count(t => t.IsCompletedSuccessfully); return new(count); } - - public record BulkCancelWorkflowInstancesRequest(ICollection Ids); - - public record BulkCancelWorkflowInstancesResponse([property: JsonPropertyName("cancelled")] int CancelledCount); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs new file mode 100644 index 000000000..3732bfeea --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs @@ -0,0 +1,44 @@ +using Elsa.Extensions; +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Notifications; +using Elsa.Workflows.Signals; + +namespace Elsa.Workflows; + +public partial class ActivityExecutionContext +{ + private readonly CancellationTokenRegistration _cancellationRegistration; + private readonly CancellationTokenSource _cancellationTokenSource; + private readonly INotificationSender _publisher; + + private void CancelActivity() + { + // If the activity is not running, do nothing. + if (Status != ActivityStatus.Running && Status != ActivityStatus.Faulted) + return; + + _ = Task.Run(async () => await CancelActivityAsync()); + } + + private async Task CancelActivityAsync() + { + // Select all child contexts. + var childContexts = WorkflowExecutionContext.ActivityExecutionContexts.Where(x => x.ParentActivityExecutionContext == this).ToList(); + + foreach (var childContext in childContexts) + childContext._cancellationTokenSource.Cancel(); + + TransitionTo(ActivityStatus.Canceled); + ClearBookmarks(); + ClearCompletionCallbacks(); + WorkflowExecutionContext.Bookmarks.RemoveWhere(x => x.ActivityNodeId == NodeId); + + // Add an execution log entry. + AddExecutionLogEntry("Canceled", payload: JournalData, includeActivityState: true); + + _cancellationRegistration.Dispose(); + + await this.SendSignalAsync(new CancelSignal()); + await _publisher.SendAsync(new ActivityCancelled(this)); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.ExecutionLogEntry.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.ExecutionLogEntry.cs new file mode 100644 index 000000000..a7ce2cffb --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.ExecutionLogEntry.cs @@ -0,0 +1,40 @@ +using Elsa.Extensions; +using Elsa.Workflows.Models; + +namespace Elsa.Workflows; + +public partial class ActivityExecutionContext +{ + /// + /// Adds a new to the execution log of the current . + /// + /// The name of the event. + /// The message of the event. + /// The source of the activity. For example, the source file name and line number in case of composite activities. + /// Any contextual data related to this event. + /// True to include activity state with this event, false otherwise. + /// Returns the created . + public WorkflowExecutionLogEntry AddExecutionLogEntry(string eventName, string? message = default, string? source = default, object? payload = default, bool includeActivityState = false) + { + var activityState = includeActivityState ? ActivityState : default; + + var logEntry = new WorkflowExecutionLogEntry( + Id, + ParentActivityExecutionContext?.Id, + Activity.Id, + Activity.Type, + Activity.Version, + Activity.Name, + NodeId, + activityState, + _systemClock.UtcNow, + WorkflowExecutionContext.ExecutionLogSequence++, + eventName, + message, + source ?? Activity.GetSource(), + payload); + + WorkflowExecutionContext.ExecutionLog.Add(logEntry); + return logEntry; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs index b53b7958a..9583fbc20 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs @@ -4,6 +4,7 @@ using Elsa.Common.Contracts; using Elsa.Expressions.Helpers; using Elsa.Expressions.Models; using Elsa.Extensions; +using Elsa.Mediator.Contracts; using Elsa.Workflows.Contracts; using Elsa.Workflows.Memory; using Elsa.Workflows.Models; @@ -15,7 +16,7 @@ namespace Elsa.Workflows; /// /// Represents the context of an activity execution. /// -public class ActivityExecutionContext : IExecutionContext +public partial class ActivityExecutionContext : IExecutionContext { private readonly ISystemClock _systemClock; private readonly List _bookmarks = new(); @@ -47,6 +48,10 @@ public class ActivityExecutionContext : IExecutionContext Tag = tag; CancellationToken = cancellationToken; Id = id; + _publisher = GetRequiredService(); + + _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + _cancellationRegistration = _cancellationTokenSource.Token.Register(CancelActivity); } /// @@ -123,8 +128,21 @@ public class ActivityExecutionContext : IExecutionContext /// /// The current status of the activity. /// - public ActivityStatus Status { get; set; } + public ActivityStatus Status { get; private set; } + /// + /// Sets the current status of the activity. + /// + public void TransitionTo(ActivityStatus status) + { + Status = status; + + if (Status is ActivityStatus.Completed + or ActivityStatus.Canceled + or ActivityStatus.Faulted) + _cancellationRegistration.Dispose(); + } + /// /// Gets or sets the exception that occurred during the activity execution, if any. /// diff --git a/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.Cancel.cs b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.Cancel.cs new file mode 100644 index 000000000..74a21f7f6 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.Cancel.cs @@ -0,0 +1,37 @@ +namespace Elsa.Workflows; + +/// +/// Provides context to the currently executing workflow. +/// +public partial class WorkflowExecutionContext +{ + private ICollection _cancellationTokenSources = new List(); + private ICollection _cancellationRegistrations = new List(); + + /// + /// Cancels the workflow and all it's children. + /// + public void Cancel() + { + foreach (var source in _cancellationTokenSources) + source.Cancel(); + + _cancellationTokenSources.Clear(); + } + + private void CancelWorkflow() + { + Bookmarks.Clear(); + _completionCallbackEntries.Clear(); + + if (Status != WorkflowStatus.Running && SubStatus != WorkflowSubStatus.Faulted) + return; + + AddExecutionLogEntry("Workflow cancelled"); + + TransitionTo(WorkflowSubStatus.Cancelled); + + foreach (var registration in _cancellationRegistrations) + registration.Dispose(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.ExecutionLogEntry.cs b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.ExecutionLogEntry.cs new file mode 100644 index 000000000..137f372fd --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.ExecutionLogEntry.cs @@ -0,0 +1,36 @@ +using Elsa.Extensions; +using Elsa.Workflows.Models; + +namespace Elsa.Workflows; + +public partial class WorkflowExecutionContext +{ + /// + /// Adds a new to the execution log of the current . + /// + /// The name of the event. + /// The message of the event. + /// Any contextual data related to this event. + /// Returns the created . + public WorkflowExecutionLogEntry AddExecutionLogEntry(string eventName, string? message = default, object? payload = default) + { + var logEntry = new WorkflowExecutionLogEntry( + Id, + default, + Workflow.Id, + Workflow.Type, + Workflow.Identity.Version, + Workflow.Name, + Workflow.Identity.Id, + default, + SystemClock.UtcNow, + ExecutionLogSequence++, + eventName, + message, + Workflow.GetSource(), + payload); + + ExecutionLog.Add(logEntry); + return logEntry; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs index 51bcc4ae3..30df7271b 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs @@ -28,7 +28,7 @@ public record ActivityCompletionCallbackEntry(ActivityExecutionContext Owner, Ac /// Provides context to the currently executing workflow. /// [PublicAPI] -public class WorkflowExecutionContext : IExecutionContext +public partial class WorkflowExecutionContext : IExecutionContext { private static readonly object ActivityOutputRegistryKey = new(); private static readonly object LastActivityResultKey = new(); @@ -37,6 +37,7 @@ public class WorkflowExecutionContext : IExecutionContext private readonly IList _completionCallbackEntries = new List(); private IList _activityExecutionContexts; private readonly IHasher _hasher; + private readonly Action? _statusUpdatedCallback; /// /// Initializes a new instance of . @@ -51,6 +52,7 @@ public class WorkflowExecutionContext : IExecutionContext string? triggerActivityId, IEnumerable incidents, DateTimeOffset createdAt, + Action? statusUpdatedCallback, CancellationTokens cancellationTokens) { ServiceProvider = serviceProvider; @@ -70,6 +72,13 @@ public class WorkflowExecutionContext : IExecutionContext CreatedAt = createdAt; CancellationTokens = cancellationTokens; Incidents = incidents.ToList(); + + var appSource = CancellationTokenSource.CreateLinkedTokenSource(CancellationTokens.ApplicationCancellationToken); + _cancellationTokenSources.Add(appSource); + var sysSource = CancellationTokenSource.CreateLinkedTokenSource(CancellationTokens.SystemCancellationToken); + _cancellationTokenSources.Add(sysSource); + _cancellationRegistrations.Add(appSource.Token.Register(CancelWorkflow)); + _cancellationRegistrations.Add(sysSource.Token.Register(CancelWorkflow)); } /// @@ -84,6 +93,7 @@ public class WorkflowExecutionContext : IExecutionContext IDictionary? properties = default, ExecuteActivityDelegate? executeDelegate = default, string? triggerActivityId = default, + Action? statusUpdatedCallback = default, CancellationTokens cancellationTokens = default) { var systemClock = serviceProvider.GetRequiredService(); @@ -99,6 +109,7 @@ public class WorkflowExecutionContext : IExecutionContext properties, executeDelegate, triggerActivityId, + statusUpdatedCallback, cancellationTokens ); } @@ -115,6 +126,7 @@ public class WorkflowExecutionContext : IExecutionContext IDictionary? properties = default, ExecuteActivityDelegate? executeDelegate = default, string? triggerActivityId = default, + Action? statusUpdatedCallback = default, CancellationTokens cancellationTokens = default) { var workflowExecutionContext = await CreateAsync( @@ -128,6 +140,7 @@ public class WorkflowExecutionContext : IExecutionContext properties, executeDelegate, triggerActivityId, + statusUpdatedCallback, cancellationTokens); var workflowStateExtractor = serviceProvider.GetRequiredService(); @@ -150,6 +163,7 @@ public class WorkflowExecutionContext : IExecutionContext IDictionary? properties = default, ExecuteActivityDelegate? executeDelegate = default, string? triggerActivityId = default, + Action? statusUpdatedCallback = default, CancellationTokens cancellationTokens = default) { // Setup a workflow execution context. @@ -163,6 +177,7 @@ public class WorkflowExecutionContext : IExecutionContext triggerActivityId, incidents, createdAt, + statusUpdatedCallback, cancellationTokens) { MemoryRegister = workflow.CreateRegister() @@ -521,6 +536,20 @@ public class WorkflowExecutionContext : IExecutionContext throw new Exception($"Cannot transition from {SubStatus} to {subStatus}"); SubStatus = subStatus; + + //For now only trigger on Cancelled, since the other statuses are handling via the host/runner + if (SubStatus == WorkflowSubStatus.Cancelled + && _statusUpdatedCallback is not null) + _statusUpdatedCallback(this); + + if (Status == WorkflowStatus.Finished + || SubStatus == WorkflowSubStatus.Suspended) + { + foreach (var registration in _cancellationRegistrations) + { + registration.Dispose(); + } + } } /// diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs index 937934592..631ba1fa6 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs @@ -1,7 +1,6 @@ using System.Linq.Expressions; using System.Reflection; using System.Text.Json; -using Elsa.Common.Contracts; using Elsa.Expressions.Contracts; using Elsa.Expressions.Helpers; using Elsa.Expressions.Models; @@ -74,45 +73,6 @@ public static class ActivityExecutionContextExtensions /// public static bool IsTriggerOfWorkflow(this ActivityExecutionContext context) => context.WorkflowExecutionContext.TriggerActivityId == context.Activity.Id; - /// - /// Adds a new to the execution log of the current . - /// - /// The being extended. - /// The name of the event. - /// The message of the event. - /// The source of the activity. For example, the source file name and line number in case of composite activities. - /// Any contextual data related to this event. - /// True to include activity state with this event, false otherwise. - /// Returns the created . - public static WorkflowExecutionLogEntry AddExecutionLogEntry(this ActivityExecutionContext context, string eventName, string? message = default, string? source = default, object? payload = default, bool includeActivityState = false) - { - var activity = context.Activity; - var activityInstanceId = context.Id; - var parentActivityInstanceId = context.ParentActivityExecutionContext?.Id; - var workflowExecutionContext = context.WorkflowExecutionContext; - var now = context.GetRequiredService().UtcNow; - var activityState = includeActivityState ? context.ActivityState : default; - - var logEntry = new WorkflowExecutionLogEntry( - activityInstanceId, - parentActivityInstanceId, - activity.Id, - activity.Type, - activity.Version, - activity.Name, - context.NodeId, - activityState, - now, - workflowExecutionContext.ExecutionLogSequence++, - eventName, - message, - source ?? activity.GetSource(), - payload); - - workflowExecutionContext.ExecutionLog.Add(logEntry); - return logEntry; - } - /// /// Creates a workflow variable by name and optionally sets the value. /// @@ -477,7 +437,7 @@ public static class ActivityExecutionContextExtensions await childContext.CancelActivityAsync(); // Mark the activity as complete. - context.Status = ActivityStatus.Completed; + context.TransitionTo(ActivityStatus.Completed); // Record the outcomes, if any. if (outcomes != null) @@ -595,7 +555,7 @@ public static class ActivityExecutionContextExtensions await CancelActivityAsync(childContext); var publisher = context.GetRequiredService(); - context.Status = ActivityStatus.Canceled; + context.TransitionTo(ActivityStatus.Canceled); context.ClearBookmarks(); context.ClearCompletionCallbacks(); context.WorkflowExecutionContext.Bookmarks.RemoveWhere(x => x.ActivityNodeId == context.NodeId); diff --git a/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs index 8e1ac8d32..e3cb8db38 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs @@ -125,36 +125,4 @@ public static class WorkflowExecutionContextExtensions /// Returns true if all activities have completed or canceled, false otherwise. /// public static bool AllActivitiesCompleted(this WorkflowExecutionContext workflowExecutionContext) => workflowExecutionContext.ActivityExecutionContexts.All(x => x.IsCompleted); - - /// - /// Adds a new to the execution log of the current . - /// - /// The being extended. - /// The name of the event. - /// The message of the event. - /// Any contextual data related to this event. - /// Returns the created . - public static WorkflowExecutionLogEntry AddExecutionLogEntry(this WorkflowExecutionContext context, string eventName, string? message = default, object? payload = default) - { - var now = context.GetRequiredService().UtcNow; - - var logEntry = new WorkflowExecutionLogEntry( - context.Id, - default, - context.Workflow.Id, - context.Workflow.Type, - context.Workflow.Identity.Version, - context.Workflow.Name, - context.Workflow.Identity.Id, - default, - now, - context.ExecutionLogSequence++, - eventName, - message, - context.Workflow.GetSource(), - payload); - - context.ExecutionLog.Add(logEntry); - return logEntry; - } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs b/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs index 32d02f4b6..fe25241c6 100644 --- a/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs +++ b/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs @@ -41,16 +41,24 @@ public class DefaultActivityInvokerMiddleware : IActivityExecutionMiddleware // Evaluate input properties. await EvaluateInputPropertiesAsync(context); - + + // Prevent the activity from being started if cancellation is requested. + if (context.CancellationToken.IsCancellationRequested) + { + context.TransitionTo(ActivityStatus.Canceled); + context.AddExecutionLogEntry("Activity cancelled"); + return; + } + // Check if the activity can be executed. if (!await context.Activity.CanExecuteAsync(context)) { - context.Status = ActivityStatus.Pending; + context.TransitionTo(ActivityStatus.Pending); context.AddExecutionLogEntry("Precondition Failed", "Cannot execute at this time"); return; } - context.Status = ActivityStatus.Running; + context.TransitionTo(ActivityStatus.Running); // Execute activity. await ExecuteActivityAsync(context); diff --git a/src/modules/Elsa.Workflows.Core/Middleware/Activities/ExceptionHandlingMiddleware.cs b/src/modules/Elsa.Workflows.Core/Middleware/Activities/ExceptionHandlingMiddleware.cs index 6a8e851ca..5d47eebc8 100644 --- a/src/modules/Elsa.Workflows.Core/Middleware/Activities/ExceptionHandlingMiddleware.cs +++ b/src/modules/Elsa.Workflows.Core/Middleware/Activities/ExceptionHandlingMiddleware.cs @@ -50,7 +50,7 @@ public class ExceptionHandlingMiddleware : IActivityExecutionMiddleware { _logger.LogWarning(e, "An exception was caught from a downstream middleware component"); context.Exception = e; - context.Status = ActivityStatus.Faulted; + context.TransitionTo(ActivityStatus.Faulted); var activity = context.Activity; var exceptionState = ExceptionState.FromException(e); diff --git a/src/modules/Elsa.Workflows.Core/Middleware/Workflows/DefaultWorkSchedulerMiddleware.cs b/src/modules/Elsa.Workflows.Core/Middleware/Workflows/DefaultActivitySchedulerMiddleware.cs similarity index 91% rename from src/modules/Elsa.Workflows.Core/Middleware/Workflows/DefaultWorkSchedulerMiddleware.cs rename to src/modules/Elsa.Workflows.Core/Middleware/Workflows/DefaultActivitySchedulerMiddleware.cs index 5d16919b4..b29e3d08e 100644 --- a/src/modules/Elsa.Workflows.Core/Middleware/Workflows/DefaultWorkSchedulerMiddleware.cs +++ b/src/modules/Elsa.Workflows.Core/Middleware/Workflows/DefaultActivitySchedulerMiddleware.cs @@ -39,6 +39,10 @@ public class DefaultActivitySchedulerMiddleware : WorkflowExecutionMiddleware while (scheduler.HasAny) { + // Do not start a workflow if cancellation has been requested. + if (context.CancellationTokens.ApplicationCancellationToken.IsCancellationRequested) + break; + var currentWorkItem = scheduler.Take(); await ExecuteWorkItemAsync(context, currentWorkItem); } diff --git a/src/modules/Elsa.Workflows.Core/Options/RunWorkflowOptions.cs b/src/modules/Elsa.Workflows.Core/Options/RunWorkflowOptions.cs index c3eb7201e..fc4a9d20c 100644 --- a/src/modules/Elsa.Workflows.Core/Options/RunWorkflowOptions.cs +++ b/src/modules/Elsa.Workflows.Core/Options/RunWorkflowOptions.cs @@ -18,4 +18,5 @@ public class RunWorkflowOptions public IDictionary? Properties { get; set; } public string? TriggerActivityId { get; set; } public CancellationTokens CancellationTokens { get; set; } + public Action? StatusUpdatedCallback { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs index 7f3ccd4e5..01f07bdda 100644 --- a/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs +++ b/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs @@ -94,6 +94,7 @@ public class WorkflowRunner : IWorkflowRunner var properties = options?.Properties; var correlationId = options?.CorrelationId; var triggerActivityId = options?.TriggerActivityId; + var statusUpdatedCallback = options?.StatusUpdatedCallback; var workflowExecutionContext = await WorkflowExecutionContext.CreateAsync( scope.ServiceProvider, workflow, @@ -103,6 +104,7 @@ public class WorkflowRunner : IWorkflowRunner properties, default, triggerActivityId, + statusUpdatedCallback, options?.CancellationTokens ?? cancellationToken); // Schedule the first activity. @@ -122,6 +124,7 @@ public class WorkflowRunner : IWorkflowRunner var properties = options?.Properties; var correlationId = options?.CorrelationId ?? workflowState.CorrelationId; var triggerActivityId = options?.TriggerActivityId; + var statusUpdatedCallback = options?.StatusUpdatedCallback; var workflowExecutionContext = await WorkflowExecutionContext.CreateAsync( scope.ServiceProvider, workflow, @@ -130,6 +133,7 @@ public class WorkflowRunner : IWorkflowRunner input, properties, default, triggerActivityId, + statusUpdatedCallback, options?.CancellationTokens ?? cancellationToken); var bookmarkId = options?.BookmarkId; diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs index aa8ef62da..15bfb6322 100644 --- a/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs +++ b/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs @@ -127,7 +127,7 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor activityExecutionContext.Id = activityExecutionContextState.Id; activityExecutionContext.Properties = properties; activityExecutionContext.ActivityState = activityExecutionContextState.ActivityState ?? new Dictionary(); - activityExecutionContext.Status = activityExecutionContextState.Status; + activityExecutionContext.TransitionTo(activityExecutionContextState.Status); activityExecutionContext.StartedAt = activityExecutionContextState.StartedAt; activityExecutionContext.CompletedAt = activityExecutionContextState.CompletedAt; activityExecutionContext.Tag = activityExecutionContextState.Tag; diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index 25578afd2..75b97539c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -102,6 +102,10 @@ public class BulkDispatchWorkflows : Activity await foreach (var item in items) { + if (context.WorkflowExecutionContext.CancellationTokens.ApplicationCancellationToken + .IsCancellationRequested) + break; + batch.Add(item); if (batch.Count < batchSize) diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkPersister.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkPersister.cs index f46df1e7c..c826d66c5 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkPersister.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkPersister.cs @@ -1,5 +1,4 @@ -using Elsa.Workflows.Helpers; -using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime.Requests; namespace Elsa.Workflows.Runtime.Contracts; @@ -11,5 +10,5 @@ public interface IBookmarksPersister /// /// Persists bookmarks and raises events. /// - Task PersistBookmarksAsync(WorkflowExecutionContext context, Diff diff); + Task PersistBookmarksAsync(UpdateBookmarksRequest updateBookmarksRequest); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionContextStore.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionContextStore.cs new file mode 100644 index 000000000..ba90ce747 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionContextStore.cs @@ -0,0 +1,24 @@ +namespace Elsa.Workflows.Runtime.Contracts; + +/// +/// Stores records. +/// +public interface IWorkflowExecutionContextStore +{ + /// + /// Saves a record of the . + /// + /// The to save. + Task SaveAsync(WorkflowExecutionContext context); + + /// + /// Finds a with the specified ID. + /// + /// The matching entity or null if no match was found. + Task FindAsync(string workflowExecutionContextId); + + /// + /// Deletes the record of the with the specified ID if it exists. + /// + Task DeleteAsync(string workflowExecutionContextId); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs index 746481379..d55af768b 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs @@ -60,6 +60,13 @@ public interface IWorkflowRuntime /// Options for executing the workflow. Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowOptions? options = default); + /// + /// Cancels the execution of a workflow. + /// + /// The ID of the workflow instance to cancel. + /// + Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default); + /// /// Finds all the workflows that can be started or resumed based on a query model. /// diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index 1f2ac45b3..23919dcde 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -73,6 +73,11 @@ public class WorkflowRuntimeFeature : FeatureBase /// public Func WorkflowInboxStore { get; set; } = sp => sp.GetRequiredService(); + /// + /// A factory that instantiates an . + /// + public Func WorkflowExecutionContextStore { get; set; } = sp => sp.GetRequiredService(); + /// /// A factory that instantiates an . /// @@ -157,6 +162,7 @@ public class WorkflowRuntimeFeature : FeatureBase .AddScoped(WorkflowExecutionLogStore) .AddScoped(ActivityExecutionLogStore) .AddScoped(WorkflowInboxStore) + .AddScoped(WorkflowExecutionContextStore) .AddSingleton(RunTaskDispatcher) .AddSingleton(BackgroundActivityScheduler) .AddScoped() @@ -188,6 +194,7 @@ public class WorkflowRuntimeFeature : FeatureBase .AddMemoryStore() .AddMemoryStore() .AddMemoryStore() + .AddMemoryStore() // Distributed locking. .AddScoped(DistributedLockProvider) @@ -207,7 +214,8 @@ public class WorkflowRuntimeFeature : FeatureBase .AddNotificationHandler() .AddNotificationHandler() .AddNotificationHandler() - .AddNotificationHandler() + .AddNotificationHandler() + .AddNotificationHandler() // Workflow activation strategies. .AddScoped() diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/WorkflowExecutionContextNotificationsHandler.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/WorkflowExecutionContextNotificationsHandler.cs new file mode 100644 index 000000000..992de6ede --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/WorkflowExecutionContextNotificationsHandler.cs @@ -0,0 +1,36 @@ +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Management.Notifications; +using Elsa.Workflows.Notifications; +using Elsa.Workflows.Runtime.Contracts; + +namespace Elsa.Workflows.Runtime.Handlers; + +/// +/// Deletes workflow execution log records in response to the notification. +/// +internal class WorkflowExecutionContextNotificationsHandler : + INotificationHandler, + INotificationHandler +{ + private readonly IWorkflowExecutionContextStore _store; + + /// + /// Initializes a new instance of the class. + /// + public WorkflowExecutionContextNotificationsHandler(IWorkflowExecutionContextStore store) + { + _store = store; + } + + /// + public Task HandleAsync(WorkflowExecuting notification, CancellationToken cancellationToken) + { + return _store.SaveAsync(notification.WorkflowExecutionContext); + } + + /// + public Task HandleAsync(WorkflowExecuted notification, CancellationToken cancellationToken) + { + return _store.DeleteAsync(notification.WorkflowExecutionContext.Id); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistBookmarkMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistBookmarkMiddleware.cs index d712632a0..12126f734 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistBookmarkMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistBookmarkMiddleware.cs @@ -1,6 +1,7 @@ using Elsa.Workflows.Helpers; using Elsa.Workflows.Pipelines.WorkflowExecution; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Requests; namespace Elsa.Workflows.Runtime.Middleware.Workflows; @@ -24,6 +25,7 @@ public class PersistBookmarkMiddleware : WorkflowExecutionMiddleware await Next(context); var updatedBookmarks = context.Bookmarks.ToList(); var diff = Diff.For(originalBookmarks, updatedBookmarks); - await _bookmarksPersister.PersistBookmarksAsync(context, diff); + var bookmarkRequest = new UpdateBookmarksRequest(context.Id, diff, context.CorrelationId); + await _bookmarksPersister.PersistBookmarksAsync(bookmarkRequest); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksIndexed.cs b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksIndexed.cs index 11cda2d6f..6303effd1 100644 --- a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksIndexed.cs +++ b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksIndexed.cs @@ -7,5 +7,5 @@ namespace Elsa.Workflows.Runtime.Notifications; /// /// The workflow execution context. /// The bookmarks that were added, removed, or unchanged. -public record WorkflowBookmarksIndexed(WorkflowExecutionContext WorkflowExecutionContext, IndexedWorkflowBookmarks IndexedWorkflowBookmarks) : INotification; +public record WorkflowBookmarksIndexed(IndexedWorkflowBookmarks IndexedWorkflowBookmarks) : INotification; diff --git a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksPersisted.cs b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksPersisted.cs index 66768f625..b5b5f2fb2 100644 --- a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksPersisted.cs +++ b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksPersisted.cs @@ -9,4 +9,4 @@ namespace Elsa.Workflows.Runtime.Notifications; /// /// The workflow execution context. /// The bookmarks that were added, removed, or unchanged. -public record WorkflowBookmarksPersisted(WorkflowExecutionContext Context, Diff Diff) : INotification; \ No newline at end of file +public record WorkflowBookmarksPersisted(Diff Diff) : INotification; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Options/StartWorkflowHostOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/StartWorkflowHostOptions.cs index 361dfc2d9..1a4697c81 100644 --- a/src/modules/Elsa.Workflows.Runtime/Options/StartWorkflowHostOptions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Options/StartWorkflowHostOptions.cs @@ -24,4 +24,9 @@ public class StartWorkflowHostOptions /// Cancellation tokens that can be used to cancel the workflow instance without cancelling system-level operations. public CancellationTokens CancellationTokens { get; set; } + + /// + /// Callback method that will be called when the status of the workflow has been updated + /// + public Action? StatusUpdatedCallback { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs index f2cc05b53..775ade777 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs @@ -1,7 +1,5 @@ using Elsa.Mediator; using Elsa.Mediator.Contracts; -using Elsa.Workflows.Helpers; -using Elsa.Workflows.Models; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Notifications; using Elsa.Workflows.Runtime.Requests; @@ -11,25 +9,15 @@ namespace Elsa.Workflows.Runtime.Services; /// public class BookmarksPersister(IBookmarkUpdater bookmarkUpdater, INotificationSender notificationSender) : IBookmarksPersister { - - /// - public async Task PersistBookmarksAsync(WorkflowExecutionContext context, Diff diff) + public async Task PersistBookmarksAsync(UpdateBookmarksRequest updateBookmarksRequest) { - var cancellationToken = context.CancellationTokens.SystemCancellationToken; - var updateBookmarksContext = new UpdateBookmarksRequest(context.Id, diff, context.CorrelationId); - await bookmarkUpdater.UpdateBookmarksAsync(updateBookmarksContext, cancellationToken); - + await bookmarkUpdater.UpdateBookmarksAsync(updateBookmarksRequest); + // Publish domain event. - await notificationSender.SendAsync(new WorkflowBookmarksIndexed(context, new IndexedWorkflowBookmarks(context.Id, diff.Added, diff.Removed, diff.Unchanged)), cancellationToken); - - // Notify all interested activities that the bookmarks have been persisted. - var activityExecutionContexts = context.ActivityExecutionContexts.Where(x => x.Activity is IBookmarksPersistedHandler && x.Bookmarks.Any()).ToList(); - - foreach (var activityExecutionContext in activityExecutionContexts) - await ((IBookmarksPersistedHandler)activityExecutionContext.Activity).BookmarksPersistedAsync(activityExecutionContext); - + await notificationSender.SendAsync(new WorkflowBookmarksIndexed(new IndexedWorkflowBookmarks(updateBookmarksRequest.WorkflowInstanceId, updateBookmarksRequest.Diff.Added, updateBookmarksRequest.Diff.Removed, updateBookmarksRequest.Diff.Unchanged))); + // Publish domain event. - await notificationSender.SendAsync(new WorkflowBookmarksPersisted(context, diff), NotificationStrategy.Background, cancellationToken); + await notificationSender.SendAsync(new WorkflowBookmarksPersisted(updateBookmarksRequest.Diff), NotificationStrategy.Background); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs index 11730d0af..8b5bf4ad8 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs @@ -1,6 +1,7 @@ using Elsa.Common.Models; using Elsa.Extensions; using Elsa.Workflows.Contracts; +using Elsa.Workflows.Helpers; using Elsa.Workflows.Management.Contracts; using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Management.Mappers; @@ -32,6 +33,10 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime private readonly IWorkflowInstanceFactory _workflowInstanceFactory; private readonly WorkflowStateMapper _workflowStateMapper; private readonly IIdentityGenerator _identityGenerator; + private readonly IWorkflowExecutionContextStore _workflowExecutionContextStore; + private readonly IWorkflowStateExtractor _workflowStateExtractor; + private readonly IServiceProvider _serviceProvider; + private readonly IBookmarksPersister _bookmarksPersister; /// /// Constructor. @@ -47,7 +52,11 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime IDistributedLockProvider distributedLockProvider, IWorkflowInstanceFactory workflowInstanceFactory, WorkflowStateMapper workflowStateMapper, - IIdentityGenerator identityGenerator) + IIdentityGenerator identityGenerator, + IWorkflowExecutionContextStore workflowExecutionContextStore, + IWorkflowStateExtractor workflowStateExtractor, + IServiceProvider serviceProvider, + IBookmarksPersister bookmarksPersister) { _workflowHostFactory = workflowHostFactory; _workflowDefinitionService = workflowDefinitionService; @@ -60,6 +69,10 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime _workflowInstanceFactory = workflowInstanceFactory; _workflowStateMapper = workflowStateMapper; _identityGenerator = identityGenerator; + _workflowExecutionContextStore = workflowExecutionContextStore; + _workflowStateExtractor = workflowStateExtractor; + _serviceProvider = serviceProvider; + _bookmarksPersister = bookmarksPersister; } /// @@ -90,6 +103,67 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime { return await StartWorkflowAsync(definitionId, options); } + + /// + public async Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken) + { + var workflowExecutionContext = await _workflowExecutionContextStore.FindAsync(workflowInstanceId); + + if (workflowExecutionContext is null) + { + // The execution context is not running on this instance. + // It might not be running on any instance, so check the db and update the record. + // Use lock to prevent race conditions and other instances from updating the workflow context + await using var cancelLock = await _distributedLockProvider.TryAcquireLockAsync($"{workflowInstanceId}-cancel"); + if (cancelLock == null) + return; + + var workflowInstance = await _workflowInstanceStore.FindAsync(workflowInstanceId, cancellationToken); + if (workflowInstance is null + || workflowInstance.SubStatus == WorkflowSubStatus.Cancelled + || workflowInstance.SubStatus == WorkflowSubStatus.Faulted) + return; + + var workflowState = await ExportWorkflowStateAsync(workflowInstanceId, cancellationToken); + + if (workflowState == null) + throw new Exception("Workflow state not found"); + + var workflowDefinition = await _workflowDefinitionService.FindAsync(workflowState.DefinitionId, VersionOptions.SpecificVersion(workflowState.DefinitionVersion), cancellationToken); + + if (workflowDefinition == null) + throw new Exception("Workflow definition not found"); + + var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken); + workflowExecutionContext = await WorkflowExecutionContext.CreateAsync(_serviceProvider, workflow, workflowState, cancellationTokens: cancellationToken); + + if (!cancellationToken.IsCancellationRequested) + await CancelWorkflowExecutionContextAsync(); + + return; + } + + await using var mainCancelLock = await _distributedLockProvider.AcquireLockAsync($"{workflowInstanceId}-cancel", TimeSpan.FromMinutes(1)); + + await CancelWorkflowExecutionContextAsync(); + + async Task CancelWorkflowExecutionContextAsync() + { + var originalBookmarks = workflowExecutionContext.Bookmarks.ToList(); + + workflowExecutionContext.Cancel(); + + var newBookmarks = workflowExecutionContext.Bookmarks.ToList(); + var diff = Diff.For(originalBookmarks, newBookmarks); + var bookmarkRequest = new UpdateBookmarksRequest(workflowExecutionContext.Id, + diff, + workflowExecutionContext.CorrelationId); + await _bookmarksPersister.PersistBookmarksAsync(bookmarkRequest); + + var instance = await _workflowInstanceManager.SaveAsync(workflowExecutionContext); + await _workflowInstanceStore.SaveAsync(instance); + } + } /// public async Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions options) diff --git a/src/modules/Elsa.Workflows.Runtime/Services/MemoryWorkflowExecutionContextStore.cs b/src/modules/Elsa.Workflows.Runtime/Services/MemoryWorkflowExecutionContextStore.cs new file mode 100644 index 000000000..7e262cd4c --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/MemoryWorkflowExecutionContextStore.cs @@ -0,0 +1,42 @@ +using Elsa.Common.Services; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Stores; + +namespace Elsa.Workflows.Runtime.Services; + +/// +/// Stores in memory. +/// +public class MemoryWorkflowExecutionContextStore : IWorkflowExecutionContextStore +{ + private readonly MemoryStore _store; + + /// + /// Initializes a new instance of the class. + /// + public MemoryWorkflowExecutionContextStore(MemoryStore store) + { + _store = store; + } + + /// + public Task SaveAsync(WorkflowExecutionContext context) + { + _store.Save(context, x => x.Id); + return Task.CompletedTask; + } + + /// + public Task FindAsync(string workflowExecutionContextId) + { + var result = _store.Find((context) => context.Id == workflowExecutionContextId); + return Task.FromResult(result); + } + + /// + public Task DeleteAsync(string workflowExecutionContextId) + { + _store.Delete(workflowExecutionContextId); + return Task.CompletedTask; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowHost.cs b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowHost.cs index 9980c8639..07eb48939 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowHost.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowHost.cs @@ -80,6 +80,7 @@ public class WorkflowHost : IWorkflowHost Input = input, Properties = properties, TriggerActivityId = options?.TriggerActivityId, + StatusUpdatedCallback = options?.StatusUpdatedCallback, CancellationTokens = options?.CancellationTokens ?? cancellationToken }; diff --git a/test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj b/test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj index 29f7310ed..43f1f116e 100644 --- a/test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj +++ b/test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj @@ -9,6 +9,7 @@ + runtime; build; native; contentfiles; analyzers; buildtransitive @@ -23,6 +24,7 @@ + diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/DefaultRuntimeTests.cs b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/DefaultRuntimeTests.cs new file mode 100644 index 000000000..7222bdece --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/DefaultRuntimeTests.cs @@ -0,0 +1,102 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Elsa.IntegrationTests.Scenarios.WorkflowCancellation.Workflows; +using Elsa.Mediator.HostedServices; +using Elsa.Mediator.Options; +using Elsa.Testing.Shared; +using Elsa.Workflows; +using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Options; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Xunit; +using Xunit.Abstractions; + +namespace Elsa.IntegrationTests.Scenarios.WorkflowCancellation; + +public class DefaultRuntimeTests +{ + private readonly IServiceProvider _services; + private readonly CapturingTextWriter _capturingTextWriter = new(); + + private readonly IWorkflowRuntime _workflowRuntime; + private readonly BackgroundCommandSenderHostedService _backgroundCommandSenderHostedService; + private readonly BackgroundEventPublisherHostedService _backgroundEventPublisherHostedService; + + public DefaultRuntimeTests(ITestOutputHelper testOutputHelper) + { + _services = new TestApplicationBuilder(testOutputHelper) + .WithCapturingTextWriter(_capturingTextWriter) + .AddWorkflow() + .AddWorkflow() + .AddWorkflow() + .AddWorkflow() + .ConfigureServices(services => + { + services + .AddSingleton(sp => + { + var options = sp.GetRequiredService>().Value; + return ActivatorUtilities.CreateInstance(sp, options.CommandWorkerCount); + }) + .AddSingleton(sp => + { + var options = sp.GetRequiredService>().Value; + return ActivatorUtilities.CreateInstance(sp, options.NotificationWorkerCount); + }) + ; + }) + .Build(); + + _backgroundCommandSenderHostedService = _services.GetRequiredService(); + _backgroundEventPublisherHostedService = _services.GetRequiredService(); + _workflowRuntime = _services.GetRequiredService(); + } + + [Fact(DisplayName = "Cancelling a suspended workflow")] + public async Task SuspendedCancelTest() + { + // Populate registries. + await _services.PopulateRegistriesAsync(); + + const string workflowDefinitionId = nameof(SimpleSuspendedWorkflow); + var workflowState = await _workflowRuntime.StartWorkflowAsync(workflowDefinitionId, new StartWorkflowRuntimeOptions()); + + Assert.Equal(WorkflowStatus.Running, workflowState.Status); + Assert.Equal(WorkflowSubStatus.Suspended, workflowState.SubStatus); + + await _workflowRuntime.CancelWorkflowAsync(workflowState.WorkflowInstanceId); + var lastWorkflowState = await _workflowRuntime.ExportWorkflowStateAsync(workflowState.WorkflowInstanceId); + + Assert.Equal(WorkflowStatus.Finished, lastWorkflowState!.Status); + Assert.Equal(WorkflowSubStatus.Cancelled, lastWorkflowState.SubStatus); + Assert.Empty(_capturingTextWriter.Lines); + } + + [Fact(DisplayName = "Cancelling a running workflow")] + public async Task RunningCancelTest() + { + await _backgroundCommandSenderHostedService.StartAsync(CancellationToken.None); + await _backgroundEventPublisherHostedService.StartAsync(CancellationToken.None); + + // Populate registries. + await _services.PopulateRegistriesAsync(); + + const string workflowDefinitionId = nameof(BulkSuspendedWorkflow); + var workflowState = await _workflowRuntime.StartWorkflowAsync(workflowDefinitionId, new StartWorkflowRuntimeOptions()); + + var bookmarks = new Stack(workflowState.Bookmarks); + var resumeOptions = new ResumeWorkflowRuntimeOptions { BookmarkId = bookmarks.Pop().Id }; + var state = await _workflowRuntime.ResumeWorkflowAsync(workflowState.WorkflowInstanceId,resumeOptions); + + await _workflowRuntime.CancelWorkflowAsync(workflowState.WorkflowInstanceId); + var lastWorkflowState = await _workflowRuntime.ExportWorkflowStateAsync(workflowState.WorkflowInstanceId); + + Assert.Equal(WorkflowStatus.Finished, lastWorkflowState!.Status); + Assert.Equal(WorkflowSubStatus.Cancelled, lastWorkflowState.SubStatus); + Assert.NotEmpty(_capturingTextWriter.Lines); + } +} \ No newline at end of file diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/ProtoActorTests.cs b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/ProtoActorTests.cs new file mode 100644 index 000000000..1a987f62a --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/ProtoActorTests.cs @@ -0,0 +1,124 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Elsa.Extensions; +using Elsa.IntegrationTests.Scenarios.WorkflowCancellation.Workflows; +using Elsa.Mediator.HostedServices; +using Elsa.Mediator.Options; +using Elsa.ProtoActor.HostedServices; +using Elsa.Testing.Shared; +using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Options; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Proto.Persistence.Sqlite; +using Xunit; +using Xunit.Abstractions; +using WorkflowStatus = Elsa.Workflows.WorkflowStatus; +using WorkflowSubStatus = Elsa.Workflows.WorkflowSubStatus; + +namespace Elsa.IntegrationTests.Scenarios.WorkflowCancellation; + +public class ProtoActorTests +{ + private readonly IServiceProvider _services; + private readonly CapturingTextWriter _capturingTextWriter = new(); + + private readonly IWorkflowRuntime _workflowRuntime; + private readonly WorkflowServerHost _workflowServerHost; + private readonly BackgroundCommandSenderHostedService _backgroundCommandSenderHostedService; + private readonly BackgroundEventPublisherHostedService _backgroundEventPublisherHostedService; + + public ProtoActorTests(ITestOutputHelper testOutputHelper) + { + _services = new TestApplicationBuilder(testOutputHelper) + .WithCapturingTextWriter(_capturingTextWriter) + .AddWorkflow() + .AddWorkflow() + .AddWorkflow() + .AddWorkflow() + .ConfigureServices(services => + { + services + .AddSingleton(sp => ActivatorUtilities.CreateInstance(sp)); + + services + .AddSingleton(sp => + { + var options = sp.GetRequiredService>().Value; + return ActivatorUtilities.CreateInstance(sp, + options.CommandWorkerCount); + }) + .AddSingleton(sp => + { + var options = sp.GetRequiredService>().Value; + return ActivatorUtilities.CreateInstance(sp, + options.NotificationWorkerCount); + }); + }).ConfigureElsa(elsa => elsa.UseWorkflowRuntime(runtime => runtime.UseProtoActor(protoActor => + { + protoActor.PersistenceProvider = _ => + new SqliteProvider( + new SqliteConnectionStringBuilder("Data Source=elsa.sqlite.db;Cache=Shared;")); + } + ))) + .Build(); + + _backgroundCommandSenderHostedService = _services.GetRequiredService(); + _backgroundEventPublisherHostedService = _services.GetRequiredService(); + _workflowServerHost = _services.GetRequiredService(); + _workflowRuntime = _services.GetRequiredService(); + } + + [Fact(DisplayName = "Cancelling a suspended workflow")] + public async Task SuspendedCancelTest() + { + // Populate registries. + await _services.PopulateRegistriesAsync(); + await _workflowServerHost.StartAsync(default); + const string workflowDefinitionId = nameof(SimpleSuspendedWorkflow); + var workflowState = + await _workflowRuntime.StartWorkflowAsync(workflowDefinitionId, new StartWorkflowRuntimeOptions()); + + Assert.Equal(WorkflowStatus.Running, workflowState.Status); + Assert.Equal(WorkflowSubStatus.Suspended, workflowState.SubStatus); + + await _workflowRuntime.CancelWorkflowAsync(workflowState.WorkflowInstanceId); + + await Task.Delay(2000); + var lastWorkflowState = await _workflowRuntime.ExportWorkflowStateAsync(workflowState.WorkflowInstanceId); + + Assert.Equal(WorkflowStatus.Finished, lastWorkflowState!.Status); + Assert.Equal(WorkflowSubStatus.Cancelled, lastWorkflowState.SubStatus); + Assert.Empty(_capturingTextWriter.Lines); + } + + [Fact(DisplayName = "Cancelling a running workflow")] + public async Task RunningCancelTest() + { + // Populate registries. + await _services.PopulateRegistriesAsync(); + await _workflowServerHost.StartAsync(default); + await _backgroundCommandSenderHostedService.StartAsync(CancellationToken.None); + await _backgroundEventPublisherHostedService.StartAsync(CancellationToken.None); + + const string workflowDefinitionId = nameof(BulkSuspendedWorkflow); + var workflowState = + await _workflowRuntime.StartWorkflowAsync(workflowDefinitionId, new StartWorkflowRuntimeOptions()); + + var bookmarks = new Stack(workflowState.Bookmarks); + var resumeOptions = new ResumeWorkflowRuntimeOptions { BookmarkId = bookmarks.Pop().Id }; + await _workflowRuntime.ResumeWorkflowAsync(workflowState.WorkflowInstanceId, resumeOptions); + + await _workflowRuntime.CancelWorkflowAsync(workflowState.WorkflowInstanceId); + var lastWorkflowState = await _workflowRuntime.ExportWorkflowStateAsync(workflowState.WorkflowInstanceId); + + Assert.Equal(WorkflowStatus.Finished, lastWorkflowState!.Status); + Assert.Equal(WorkflowSubStatus.Cancelled, lastWorkflowState.SubStatus); + + Assert.NotEmpty(_capturingTextWriter.Lines); + } +} \ No newline at end of file diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/BulkSuspendedWorkflow.cs b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/BulkSuspendedWorkflow.cs new file mode 100644 index 000000000..afb88519b --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/BulkSuspendedWorkflow.cs @@ -0,0 +1,33 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using Elsa.Scheduling.Activities; +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Contracts; +using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime.Activities; + +namespace Elsa.IntegrationTests.Scenarios.WorkflowCancellation.Workflows; + +public class BulkSuspendedWorkflow : WorkflowBase +{ + protected override void Build(IWorkflowBuilder builder) + { + object[] items = Enumerable.Range(0,10000).Select(x => (object) x).ToArray(); + + builder.Root = new Sequence + { + Activities = + { + new Start(), + new Delay(TimeSpan.FromSeconds(10)), + new BulkDispatchWorkflows + { + WorkflowDefinitionId = new Input(nameof(SimpleChildWorkflow)), + Items = new Input>(items) + } + }, + }; + } +} \ No newline at end of file diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/ResumeDispatchWorkflow.cs b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/ResumeDispatchWorkflow.cs new file mode 100644 index 000000000..8292b7026 --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/ResumeDispatchWorkflow.cs @@ -0,0 +1,21 @@ +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Contracts; +using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime.Activities; + +namespace Elsa.IntegrationTests.Scenarios.WorkflowCancellation.Workflows; + +public class ResumeDispatchWorkflow : WorkflowBase +{ + protected override void Build(IWorkflowBuilder builder) + { + builder.Root = new Sequence + { + Activities = + { + new PublishEvent { EventName = new Input("ResumeBlockDispatch") } + } + }; + } +} \ No newline at end of file diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleChildWorkflow.cs b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleChildWorkflow.cs new file mode 100644 index 000000000..b8217afaf --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleChildWorkflow.cs @@ -0,0 +1,19 @@ +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Contracts; + +namespace Elsa.IntegrationTests.Scenarios.WorkflowCancellation.Workflows; + +public class SimpleChildWorkflow : WorkflowBase +{ + protected override void Build(IWorkflowBuilder builder) + { + builder.Root = new Sequence + { + Activities = + { + new WriteLine("Executed SimpleChildWorkflow") + } + }; + } +} \ No newline at end of file diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleSuspendedWorkflow.cs b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleSuspendedWorkflow.cs new file mode 100644 index 000000000..7b0912b6d --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleSuspendedWorkflow.cs @@ -0,0 +1,22 @@ +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Contracts; +using Elsa.Workflows.Runtime.Activities; + +namespace Elsa.IntegrationTests.Scenarios.WorkflowCancellation.Workflows; + +public class SimpleSuspendedWorkflow : WorkflowBase +{ + protected override void Build(IWorkflowBuilder builder) + { + builder.Root = new Sequence + { + Activities = + { + new Start(), + new Event("BlockingEvent"), + new WriteLine("Workflow was not properly blocked") + }, + }; + } +} \ No newline at end of file