SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Workflow cancellation (#4813)

Browse source 
* 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 <sipkeschoorstra@outlook.com>
This commit is contained in:
raymonddenhaan 2024-01-23 12:05:03 +01:00 committed by GitHub
parent 832fac4db1
commit c409414152
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
42 changed files with 854 additions and 122 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Elsa.sln
View file

@ -1,4 +1,4 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 17
VisualStudioVersion = 17.7.34003.232

2
src/modules/Elsa.Alterations/AlterationHandlers/ScheduleActivityHandler.cs
View file

@ -28,7 +28,7 @@ public class ScheduleActivityHandler : AlterationHandlerBase<ScheduleActivity>
{
// 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;

60
src/modules/Elsa.ProtoActor/Grains/WorkflowInstance.cs
View file

@ -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<CancellationTokenSource> _cancellationTokenSources = new List<CancellationTokenSource>();
/// <inheritdoc />
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<IWorkflowStateExtractor>();
var bookmarkPersistor = scope.ServiceProvider.GetRequiredService<IBookmarksPersister>();
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);
}
/// <inheritdoc />
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
/// <inheritdoc />
public override Task<WorkflowExecutionResponse> 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();
}
/// <inheritdoc />
public override async Task<ExportWorkflowStateResponse> ExportState(ExportWorkflowStateRequest request)
{

1
src/modules/Elsa.ProtoActor/Proto/WorkflowInstance.proto
View file

@ -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);
}

10
src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs
View file

@ -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!;
}
/// <inheritdoc />
public async Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken)
{
var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId);
await client.Cancel(cancellationToken);
}
/// <inheritdoc />
public async Task<IEnumerable<WorkflowMatch>> 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);
}

2
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDelete/Endpoint.cs
View file

@ -22,7 +22,7 @@ internal class BulkDelete : ElsaEndpoint<Request, Response>
public override async Task<Response> 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);
}
}

21
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/BulkCancel/Endpoint.cs
View file

@ -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<Request, Response>
{
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<Response> 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<string> Ids);
public record BulkCancelWorkflowInstancesResponse([property: JsonPropertyName("cancelled")] int CancelledCount);
}

44
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs Normal file
View file

@ -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));
}
}

40
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.ExecutionLogEntry.cs Normal file
View file

@ -0,0 +1,40 @@
using Elsa.Extensions;
using Elsa.Workflows.Models;
namespace Elsa.Workflows;
public partial class ActivityExecutionContext
{
/// <summary>
/// Adds a new <see cref="WorkflowExecutionLogEntry"/> to the execution log of the current <see cref="Workflows.WorkflowExecutionContext"/>.
/// </summary>
/// <param name="eventName">The name of the event.</param>
/// <param name="message">The message of the event.</param>
/// <param name="source">The source of the activity. For example, the source file name and line number in case of composite activities.</param>
/// <param name="payload">Any contextual data related to this event.</param>
/// <param name="includeActivityState">True to include activity state with this event, false otherwise.</param>
/// <returns>Returns the created <see cref="WorkflowExecutionLogEntry"/>.</returns>
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;
}
}

22
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs
View file

@ -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;
/// <summary>
/// Represents the context of an activity execution.
/// </summary>
public class ActivityExecutionContext : IExecutionContext
public partial class ActivityExecutionContext : IExecutionContext
{
private readonly ISystemClock _systemClock;
private readonly List<Bookmark> _bookmarks = new();
@ -47,6 +48,10 @@ public class ActivityExecutionContext : IExecutionContext
Tag = tag;
CancellationToken = cancellationToken;
Id = id;
_publisher = GetRequiredService<INotificationSender>();
_cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
_cancellationRegistration = _cancellationTokenSource.Token.Register(CancelActivity);
}
/// <summary>
@ -123,8 +128,21 @@ public class ActivityExecutionContext : IExecutionContext
/// <summary>
/// The current status of the activity.
/// </summary>
public ActivityStatus Status { get; set; }
public ActivityStatus Status { get; private set; }
/// <summary>
/// Sets the current status of the activity.
/// </summary>
public void TransitionTo(ActivityStatus status)
{
Status = status;
if (Status is ActivityStatus.Completed
or ActivityStatus.Canceled
or ActivityStatus.Faulted)
_cancellationRegistration.Dispose();
}
/// <summary>
/// Gets or sets the exception that occurred during the activity execution, if any.
/// </summary>

37
src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.Cancel.cs Normal file
View file

@ -0,0 +1,37 @@
namespace Elsa.Workflows;
/// <summary>
/// Provides context to the currently executing workflow.
/// </summary>
public partial class WorkflowExecutionContext
{
private ICollection<CancellationTokenSource> _cancellationTokenSources = new List<CancellationTokenSource>();
private ICollection<CancellationTokenRegistration> _cancellationRegistrations = new List<CancellationTokenRegistration>();
/// <summary>
/// Cancels the workflow and all it's children.
/// </summary>
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();
}
}

36
src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.ExecutionLogEntry.cs Normal file
View file

@ -0,0 +1,36 @@
using Elsa.Extensions;
using Elsa.Workflows.Models;
namespace Elsa.Workflows;
public partial class WorkflowExecutionContext
{
/// <summary>
/// Adds a new <see cref="WorkflowExecutionLogEntry"/> to the execution log of the current <see cref="Workflows.WorkflowExecutionContext"/>.
/// </summary>
/// <param name="eventName">The name of the event.</param>
/// <param name="message">The message of the event.</param>
/// <param name="payload">Any contextual data related to this event.</param>
/// <returns>Returns the created <see cref="WorkflowExecutionLogEntry"/>.</returns>
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;
}
}

31
src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs
View file

@ -28,7 +28,7 @@ public record ActivityCompletionCallbackEntry(ActivityExecutionContext Owner, Ac
/// Provides context to the currently executing workflow.
/// </summary>
[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<ActivityCompletionCallbackEntry> _completionCallbackEntries = new List<ActivityCompletionCallbackEntry>();
private IList<ActivityExecutionContext> _activityExecutionContexts;
private readonly IHasher _hasher;
private readonly Action<WorkflowExecutionContext>? _statusUpdatedCallback;
/// <summary>
/// Initializes a new instance of <see cref="WorkflowExecutionContext"/>.
@ -51,6 +52,7 @@ public class WorkflowExecutionContext : IExecutionContext
string? triggerActivityId,
IEnumerable<ActivityIncident> incidents,
DateTimeOffset createdAt,
Action<WorkflowExecutionContext>? 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));
}
/// <summary>
@ -84,6 +93,7 @@ public class WorkflowExecutionContext : IExecutionContext
IDictionary<string, object>? properties = default,
ExecuteActivityDelegate? executeDelegate = default,
string? triggerActivityId = default,
Action<WorkflowExecutionContext>? statusUpdatedCallback = default,
CancellationTokens cancellationTokens = default)
{
var systemClock = serviceProvider.GetRequiredService<ISystemClock>();
@ -99,6 +109,7 @@ public class WorkflowExecutionContext : IExecutionContext
properties,
executeDelegate,
triggerActivityId,
statusUpdatedCallback,
cancellationTokens
);
}
@ -115,6 +126,7 @@ public class WorkflowExecutionContext : IExecutionContext
IDictionary<string, object>? properties = default,
ExecuteActivityDelegate? executeDelegate = default,
string? triggerActivityId = default,
Action<WorkflowExecutionContext>? 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<IWorkflowStateExtractor>();
@ -150,6 +163,7 @@ public class WorkflowExecutionContext : IExecutionContext
IDictionary<string, object>? properties = default,
ExecuteActivityDelegate? executeDelegate = default,
string? triggerActivityId = default,
Action<WorkflowExecutionContext>? 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();
}
}
}
/// <summary>

44
src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs
View file

@ -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
/// </summary>
public static bool IsTriggerOfWorkflow(this ActivityExecutionContext context) => context.WorkflowExecutionContext.TriggerActivityId == context.Activity.Id;
/// <summary>
/// Adds a new <see cref="WorkflowExecutionLogEntry"/> to the execution log of the current <see cref="WorkflowExecutionContext"/>.
/// </summary>
/// <param name="context">The <see cref="ActivityExecutionContext"/></param> being extended.
/// <param name="eventName">The name of the event.</param>
/// <param name="message">The message of the event.</param>
/// <param name="source">The source of the activity. For example, the source file name and line number in case of composite activities.</param>
/// <param name="payload">Any contextual data related to this event.</param>
/// <param name="includeActivityState">True to include activity state with this event, false otherwise.</param>
/// <returns>Returns the created <see cref="WorkflowExecutionLogEntry"/>.</returns>
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<ISystemClock>().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;
}
/// <summary>
/// Creates a workflow variable by name and optionally sets the value.
/// </summary>
@ -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<INotificationSender>();
context.Status = ActivityStatus.Canceled;
context.TransitionTo(ActivityStatus.Canceled);
context.ClearBookmarks();
context.ClearCompletionCallbacks();
context.WorkflowExecutionContext.Bookmarks.RemoveWhere(x => x.ActivityNodeId == context.NodeId);

32
src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs
View file

@ -125,36 +125,4 @@ public static class WorkflowExecutionContextExtensions
/// Returns true if all activities have completed or canceled, false otherwise.
/// </summary>
public static bool AllActivitiesCompleted(this WorkflowExecutionContext workflowExecutionContext) => workflowExecutionContext.ActivityExecutionContexts.All(x => x.IsCompleted);
/// <summary>
/// Adds a new <see cref="WorkflowExecutionLogEntry"/> to the execution log of the current <see cref="WorkflowExecutionContext"/>.
/// </summary>
/// <param name="context">The <see cref="WorkflowExecutionContext"/></param> being extended.
/// <param name="eventName">The name of the event.</param>
/// <param name="message">The message of the event.</param>
/// <param name="payload">Any contextual data related to this event.</param>
/// <returns>Returns the created <see cref="WorkflowExecutionLogEntry"/>.</returns>
public static WorkflowExecutionLogEntry AddExecutionLogEntry(this WorkflowExecutionContext context, string eventName, string? message = default, object? payload = default)
{
var now = context.GetRequiredService<ISystemClock>().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;
}
}

14
src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs
View file

@ -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);

2
src/modules/Elsa.Workflows.Core/Middleware/Activities/ExceptionHandlingMiddleware.cs
View file

@ -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);

4
src/modules/Elsa.Workflows.Core/Middleware/Workflows/DefaultWorkSchedulerMiddleware.cs → src/modules/Elsa.Workflows.Core/Middleware/Workflows/DefaultActivitySchedulerMiddleware.cs
View file

@ -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);
}

1
src/modules/Elsa.Workflows.Core/Options/RunWorkflowOptions.cs
View file

@ -18,4 +18,5 @@ public class RunWorkflowOptions
public IDictionary<string, object>? Properties { get; set; }
public string? TriggerActivityId { get; set; }
public CancellationTokens CancellationTokens { get; set; }
public Action<WorkflowExecutionContext>? StatusUpdatedCallback { get; set; }
}

4
src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs
View file

@ -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;

2
src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs
View file

@ -127,7 +127,7 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor
activityExecutionContext.Id = activityExecutionContextState.Id;
activityExecutionContext.Properties = properties;
activityExecutionContext.ActivityState = activityExecutionContextState.ActivityState ?? new Dictionary<string, object>();
activityExecutionContext.Status = activityExecutionContextState.Status;
activityExecutionContext.TransitionTo(activityExecutionContextState.Status);
activityExecutionContext.StartedAt = activityExecutionContextState.StartedAt;
activityExecutionContext.CompletedAt = activityExecutionContextState.CompletedAt;
activityExecutionContext.Tag = activityExecutionContextState.Tag;

4
src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs
View file

@ -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)

5
src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkPersister.cs
View file

@ -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
/// <summary>
/// Persists bookmarks and raises events.
/// </summary>
Task PersistBookmarksAsync(WorkflowExecutionContext context, Diff<Bookmark> diff);
Task PersistBookmarksAsync(UpdateBookmarksRequest updateBookmarksRequest);
}

24
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowExecutionContextStore.cs Normal file
View file

@ -0,0 +1,24 @@
namespace Elsa.Workflows.Runtime.Contracts;
/// <summary>
/// Stores <see cref="WorkflowExecutionContext"/> records.
/// </summary>
public interface IWorkflowExecutionContextStore
{
/// <summary>
/// Saves a record of the <see cref="WorkflowExecutionContext"/>.
/// </summary>
/// <param name="context">The <see cref="WorkflowExecutionContext"/> to save.</param>
Task SaveAsync(WorkflowExecutionContext context);
/// <summary>
/// Finds a <see cref="WorkflowExecutionContext"/> with the specified ID.
/// </summary>
/// <returns>The matching entity or null if no match was found.</returns>
Task<WorkflowExecutionContext?> FindAsync(string workflowExecutionContextId);
/// <summary>
/// Deletes the record of the <see cref="WorkflowExecutionContext"/> with the specified ID if it exists.
/// </summary>
Task DeleteAsync(string workflowExecutionContextId);
}

7
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs
View file

@ -60,6 +60,13 @@ public interface IWorkflowRuntime
/// <param name="options">Options for executing the workflow.</param>
Task<WorkflowExecutionResult> ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowOptions? options = default);
/// <summary>
/// Cancels the execution of a workflow.
/// </summary>
/// <param name="workflowInstanceId">The ID of the workflow instance to cancel.</param>
/// <param name="cancellationToken"></param>
Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default);
/// <summary>
/// Finds all the workflows that can be started or resumed based on a query model.
/// </summary>

10
src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs
View file

@ -73,6 +73,11 @@ public class WorkflowRuntimeFeature : FeatureBase
/// </summary>
public Func<IServiceProvider, IWorkflowInboxMessageStore> WorkflowInboxStore { get; set; } = sp => sp.GetRequiredService<MemoryWorkflowInboxMessageStore>();
/// <summary>
/// A factory that instantiates an <see cref="IWorkflowExecutionContextStore"/>.
/// </summary>
public Func<IServiceProvider, IWorkflowExecutionContextStore> WorkflowExecutionContextStore { get; set; } = sp => sp.GetRequiredService<MemoryWorkflowExecutionContextStore>();
/// <summary>
/// A factory that instantiates an <see cref="IDistributedLockProvider"/>.
/// </summary>
@ -157,6 +162,7 @@ public class WorkflowRuntimeFeature : FeatureBase
.AddScoped(WorkflowExecutionLogStore)
.AddScoped(ActivityExecutionLogStore)
.AddScoped(WorkflowInboxStore)
.AddScoped(WorkflowExecutionContextStore)
.AddSingleton(RunTaskDispatcher)
.AddSingleton(BackgroundActivityScheduler)
.AddScoped<IBookmarkManager, DefaultBookmarkManager>()
@ -188,6 +194,7 @@ public class WorkflowRuntimeFeature : FeatureBase
.AddMemoryStore<WorkflowExecutionLogRecord, MemoryWorkflowExecutionLogStore>()
.AddMemoryStore<ActivityExecutionRecord, MemoryActivityExecutionStore>()
.AddMemoryStore<WorkflowInboxMessage, MemoryWorkflowInboxMessageStore>()
.AddMemoryStore<WorkflowExecutionContext, MemoryWorkflowExecutionContextStore>()
// Distributed locking.
.AddScoped(DistributedLockProvider)
@ -207,7 +214,8 @@ public class WorkflowRuntimeFeature : FeatureBase
.AddNotificationHandler<DeleteActivityExecutionLogRecords>()
.AddNotificationHandler<ReadWorkflowInboxMessage>()
.AddNotificationHandler<DeliverWorkflowMessagesFromInbox>()
.AddNotificationHandler<DeleteWorkflowExecutionLogRecords>()
.AddNotificationHandler<DeleteWorkflowExecutionLogRecords>()
.AddNotificationHandler<WorkflowExecutionContextNotificationsHandler>()
// Workflow activation strategies.
.AddScoped<IWorkflowActivationStrategy, SingletonStrategy>()

36
src/modules/Elsa.Workflows.Runtime/Handlers/WorkflowExecutionContextNotificationsHandler.cs Normal file
View file

@ -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;
/// <summary>
/// Deletes workflow execution log records in response to the <see cref="WorkflowInstancesDeleting"/> notification.
/// </summary>
internal class WorkflowExecutionContextNotificationsHandler :
INotificationHandler<WorkflowExecuting>,
INotificationHandler<WorkflowExecuted>
{
private readonly IWorkflowExecutionContextStore _store;
/// <summary>
/// Initializes a new instance of the <see cref="DeleteWorkflowExecutionLogRecords"/> class.
/// </summary>
public WorkflowExecutionContextNotificationsHandler(IWorkflowExecutionContextStore store)
{
_store = store;
}
/// <inheritdoc />
public Task HandleAsync(WorkflowExecuting notification, CancellationToken cancellationToken)
{
return _store.SaveAsync(notification.WorkflowExecutionContext);
}
/// <inheritdoc />
public Task HandleAsync(WorkflowExecuted notification, CancellationToken cancellationToken)
{
return _store.DeleteAsync(notification.WorkflowExecutionContext.Id);
}
}

4
src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistBookmarkMiddleware.cs
View file

@ -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);
}
}

2
src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksIndexed.cs
View file

@ -7,5 +7,5 @@ namespace Elsa.Workflows.Runtime.Notifications;
/// </summary>
/// <param name="WorkflowExecutionContext">The workflow execution context.</param>
/// <param name="IndexedWorkflowBookmarks">The bookmarks that were added, removed, or unchanged.</param>
public record WorkflowBookmarksIndexed(WorkflowExecutionContext WorkflowExecutionContext, IndexedWorkflowBookmarks IndexedWorkflowBookmarks) : INotification;
public record WorkflowBookmarksIndexed(IndexedWorkflowBookmarks IndexedWorkflowBookmarks) : INotification;

2
src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowBookmarksPersisted.cs
View file

@ -9,4 +9,4 @@ namespace Elsa.Workflows.Runtime.Notifications;
/// </summary>
/// <param name="Context">The workflow execution context.</param>
/// <param name="Diff">The bookmarks that were added, removed, or unchanged.</param>
public record WorkflowBookmarksPersisted(WorkflowExecutionContext Context, Diff<Bookmark> Diff) : INotification;
public record WorkflowBookmarksPersisted(Diff<Bookmark> Diff) : INotification;

5
src/modules/Elsa.Workflows.Runtime/Options/StartWorkflowHostOptions.cs
View file

@ -24,4 +24,9 @@ public class StartWorkflowHostOptions
/// <summary>Cancellation tokens that can be used to cancel the workflow instance without cancelling system-level operations.</summary>
public CancellationTokens CancellationTokens { get; set; }
/// <summary>
/// Callback method that will be called when the status of the workflow has been updated
/// </summary>
public Action<WorkflowExecutionContext>? StatusUpdatedCallback { get; set; }
}

24
src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs
View file

@ -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;
/// <inheritdoc />
public class BookmarksPersister(IBookmarkUpdater bookmarkUpdater, INotificationSender notificationSender) : IBookmarksPersister
{
/// <inheritdoc />
public async Task PersistBookmarksAsync(WorkflowExecutionContext context, Diff<Bookmark> 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);
}
}

76
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs
View file

@ -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;
/// <summary>
/// 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;
}
/// <inheritdoc />
@ -90,6 +103,67 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
{
return await StartWorkflowAsync(definitionId, options);
}
/// <inheritdoc />
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);
}
}
/// <inheritdoc />
public async Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions options)

42
src/modules/Elsa.Workflows.Runtime/Services/MemoryWorkflowExecutionContextStore.cs Normal file
View file

@ -0,0 +1,42 @@
using Elsa.Common.Services;
using Elsa.Workflows.Runtime.Contracts;
using Elsa.Workflows.Runtime.Stores;
namespace Elsa.Workflows.Runtime.Services;
/// <summary>
/// Stores <see cref="WorkflowExecutionContext"/> in memory.
/// </summary>
public class MemoryWorkflowExecutionContextStore : IWorkflowExecutionContextStore
{
private readonly MemoryStore<WorkflowExecutionContext> _store;
/// <summary>
/// Initializes a new instance of the <see cref="MemoryActivityExecutionStore"/> class.
/// </summary>
public MemoryWorkflowExecutionContextStore(MemoryStore<WorkflowExecutionContext> store)
{
_store = store;
}
/// <inheritdoc />
public Task SaveAsync(WorkflowExecutionContext context)
{
_store.Save(context, x => x.Id);
return Task.CompletedTask;
}
/// <inheritdoc />
public Task<WorkflowExecutionContext?> FindAsync(string workflowExecutionContextId)
{
var result = _store.Find((context) => context.Id == workflowExecutionContextId);
return Task.FromResult(result);
}
/// <inheritdoc />
public Task DeleteAsync(string workflowExecutionContextId)
{
_store.Delete(workflowExecutionContextId);
return Task.CompletedTask;
}
}

1
src/modules/Elsa.Workflows.Runtime/Services/WorkflowHost.cs
View file

@ -80,6 +80,7 @@ public class WorkflowHost : IWorkflowHost
Input = input,
Properties = properties,
TriggerActivityId = options?.TriggerActivityId,
StatusUpdatedCallback = options?.StatusUpdatedCallback,
CancellationTokens = options?.CancellationTokens ?? cancellationToken
};

2
test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj
View file

@ -9,6 +9,7 @@
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Logging.Console" Version="8.0.0"/>
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.8.0"/>
<PackageReference Include="Proto.Persistence.Sqlite" Version="1.4.0" />
<PackageReference Include="xunit" Version="2.6.2"/>
<PackageReference Include="xunit.runner.visualstudio" Version="2.5.4">
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
@ -23,6 +24,7 @@
<ItemGroup>
<ProjectReference Include="..\..\..\src\common\Elsa.Testing.Shared\Elsa.Testing.Shared.csproj"/>
<ProjectReference Include="..\..\..\src\modules\Elsa.JavaScript\Elsa.JavaScript.csproj"/>
<ProjectReference Include="..\..\..\src\modules\Elsa.ProtoActor\Elsa.ProtoActor.csproj" />
<ProjectReference Include="..\..\..\src\modules\Elsa.Scheduling\Elsa.Scheduling.csproj"/>
<ProjectReference Include="..\..\..\src\modules\Elsa.WorkflowProviders.BlobStorage\Elsa.WorkflowProviders.BlobStorage.csproj"/>
<ProjectReference Include="..\..\..\src\modules\Elsa.Workflows.Api\Elsa.Workflows.Api.csproj"/>

102
test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/DefaultRuntimeTests.cs Normal file
View file

@ -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<BulkSuspendedWorkflow>()
.AddWorkflow<ResumeDispatchWorkflow>()
.AddWorkflow<SimpleChildWorkflow>()
.AddWorkflow<SimpleSuspendedWorkflow>()
.ConfigureServices(services =>
{
services
.AddSingleton(sp =>
{
var options = sp.GetRequiredService<IOptions<MediatorOptions>>().Value;
return ActivatorUtilities.CreateInstance<BackgroundCommandSenderHostedService>(sp, options.CommandWorkerCount);
})
.AddSingleton(sp =>
{
var options = sp.GetRequiredService<IOptions<MediatorOptions>>().Value;
return ActivatorUtilities.CreateInstance<BackgroundEventPublisherHostedService>(sp, options.NotificationWorkerCount);
})
;
})
.Build();
_backgroundCommandSenderHostedService = _services.GetRequiredService<BackgroundCommandSenderHostedService>();
_backgroundEventPublisherHostedService = _services.GetRequiredService<BackgroundEventPublisherHostedService>();
_workflowRuntime = _services.GetRequiredService<IWorkflowRuntime>();
}
[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<Bookmark>(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);
}
}

124
test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/ProtoActorTests.cs Normal file
View file

@ -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<BulkSuspendedWorkflow>()
.AddWorkflow<ResumeDispatchWorkflow>()
.AddWorkflow<SimpleChildWorkflow>()
.AddWorkflow<SimpleSuspendedWorkflow>()
.ConfigureServices(services =>
{
services
.AddSingleton(sp => ActivatorUtilities.CreateInstance<WorkflowServerHost>(sp));
services
.AddSingleton(sp =>
{
var options = sp.GetRequiredService<IOptions<MediatorOptions>>().Value;
return ActivatorUtilities.CreateInstance<BackgroundCommandSenderHostedService>(sp,
options.CommandWorkerCount);
})
.AddSingleton(sp =>
{
var options = sp.GetRequiredService<IOptions<MediatorOptions>>().Value;
return ActivatorUtilities.CreateInstance<BackgroundEventPublisherHostedService>(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<BackgroundCommandSenderHostedService>();
_backgroundEventPublisherHostedService = _services.GetRequiredService<BackgroundEventPublisherHostedService>();
_workflowServerHost = _services.GetRequiredService<WorkflowServerHost>();
_workflowRuntime = _services.GetRequiredService<IWorkflowRuntime>();
}
[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<Bookmark>(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);
}
}

33
test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/BulkSuspendedWorkflow.cs Normal file
View file

@ -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<string>(nameof(SimpleChildWorkflow)),
Items = new Input<ICollection<object>>(items)
}
},
};
}
}

21
test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/ResumeDispatchWorkflow.cs Normal file
View file

@ -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<string>("ResumeBlockDispatch") }
}
};
}
}

19
test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleChildWorkflow.cs Normal file
View file

@ -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")
}
};
}
}

22
test/integration/Elsa.IntegrationTests/Scenarios/WorkflowCancellation/Workflows/SimpleSuspendedWorkflow.cs Normal file
View file

@ -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")
},
};
}
}