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

Add support for resuming workflows using activity instance ID and hash

Browse source
This commit is contained in:
Sipke Schoorstra 2023-04-06 14:59:30 +02:00
parent 41e1a91748
commit cea4c8fa0a
19 changed files with 185 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs
View file

@ -7,9 +7,9 @@ namespace Elsa.MassTransit.Consumers;
/// <summary>
/// A consumer of various dispatch message types to asynchronously execute workflows.
/// </summary>
public class DispatchWorkflowRequestConsumer :
IConsumer<DispatchWorkflowDefinition>,
IConsumer<DispatchWorkflowInstance>,
public class DispatchWorkflowRequestConsumer :
IConsumer<DispatchWorkflowDefinition>,
IConsumer<DispatchWorkflowInstance>,
IConsumer<DispatchTriggerWorkflows>,
IConsumer<DispatchResumeWorkflows>
{
@ -28,7 +28,7 @@ public class DispatchWorkflowRequestConsumer :
{
var message = context.Message;
var options = new StartWorkflowRuntimeOptions(message.CorrelationId, message.Input, message.VersionOptions, InstanceId: message.InstanceId);
await _workflowRuntime.StartWorkflowAsync(message.DefinitionId, options, context.CancellationToken);
}
@ -36,7 +36,17 @@ public class DispatchWorkflowRequestConsumer :
public async Task Consume(ConsumeContext<DispatchWorkflowInstance> context)
{
var message = context.Message;
var options = new ResumeWorkflowRuntimeOptions(message.CorrelationId, message.InstanceId, message.BookmarkId, message.ActivityId, message.Input);
var options = new ResumeWorkflowRuntimeOptions(
message.CorrelationId,
message.InstanceId,
message.BookmarkId,
message.ActivityId,
message.ActivityNodeId,
message.ActivityInstanceId,
message.ActivityHash,
message.Input);
await _workflowRuntime.ResumeWorkflowAsync(message.InstanceId, options, context.CancellationToken);
}

3
src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs
View file

@ -40,6 +40,9 @@ public class MassTransitWorkflowDispatcher : IWorkflowDispatcher
request.InstanceId,
request.BookmarkId,
request.ActivityId,
request.ActivityNodeId,
request.ActivityInstanceId,
request.ActivityHash,
request.Input,
request.CorrelationId
), cancellationToken);

3
src/modules/Elsa.MassTransit/Messages/DispatchWorkflowInstance.cs
View file

@ -5,6 +5,9 @@ public record DispatchWorkflowInstance
string InstanceId,
string? BookmarkId,
string? ActivityId,
string? ActivityNodeId,
string? ActivityInstanceId,
string? ActivityHash,
IDictionary<string, object>? Input,
string? CorrelationId
);

14
src/modules/Elsa.ProtoActor/Grains/WorkflowGrain.cs
View file

@ -146,8 +146,20 @@ public class WorkflowGrain : WorkflowGrainBase
var correlationId = request.CorrelationId;
var bookmarkId = request.BookmarkId.NullIfEmpty();
var activityId = request.ActivityId.NullIfEmpty();
var activityNodeId = request.ActivityNodeId.NullIfEmpty();
var activityInstanceId = request.ActivityInstanceId.NullIfEmpty();
var activityHash = request.ActivityHash.NullIfEmpty();
var cancellationToken = Context.CancellationToken;
var resumeWorkflowHostOptions = new ResumeWorkflowHostOptions(correlationId, bookmarkId, activityId, _input);
var resumeWorkflowHostOptions = new ResumeWorkflowHostOptions(
correlationId,
bookmarkId,
activityId,
activityNodeId,
activityInstanceId,
activityHash,
_input);
var definitionId = _definitionId;
var versionOptions = VersionOptions.SpecificVersion(_version);

5
src/modules/Elsa.ProtoActor/Protos/Messages.proto
View file

@ -37,7 +37,10 @@ message ResumeWorkflowRequest {
optional string CorrelationId = 2;
optional string BookmarkId = 3;
optional string ActivityId = 4;
optional Input input = 5;
optional string ActivityNodeId = 5;
optional string ActivityInstanceId = 6;
optional string ActivityHash = 7;
optional Input input = 8;
}
message ResumeWorkflowResponse {

5
src/modules/Elsa.Workflows.Core/Contracts/IWorkflowRunner.cs
View file

@ -23,6 +23,9 @@ public record RunWorkflowOptions(
string? InstanceId = default,
string? CorrelationId = default,
string? BookmarkId = default,
string? ActivityNodeId = default,
string? ActivityId = default,
string? ActivityNodeId = default,
string? ActivityInstanceId = default,
string? ActivityHash = default,
IDictionary<string, object>? Input = default,
string? TriggerActivityId = default);

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

@ -54,6 +54,16 @@ public static class WorkflowExecutionContextExtensions
var workItem = new ActivityWorkItem(activity.Id, async () => await activityInvoker.InvokeAsync(workflowExecutionContext, activity));
workflowExecutionContext.Scheduler.Schedule(workItem);
}
/// <summary>
/// Schedules the specified activity execution context of the workflow.
/// </summary>
public static void ScheduleActivityExecutionContext(this WorkflowExecutionContext workflowExecutionContext, ActivityExecutionContext activityExecutionContext)
{
var activityInvoker = workflowExecutionContext.GetRequiredService<IActivityInvoker>();
var workItem = new ActivityWorkItem(activityExecutionContext.Activity.Id, async () => await activityInvoker.InvokeAsync(activityExecutionContext));
workflowExecutionContext.Scheduler.Schedule(workItem);
}
/// <summary>
/// Schedules the activity of the specified bookmark.

41
src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs
View file

@ -23,6 +23,7 @@ public class WorkflowExecutionContext : IExecutionContext
{
internal static ValueTask Complete(ActivityExecutionContext context) => context.CompleteActivityAsync();
private readonly IServiceProvider _serviceProvider;
private readonly IHasher _hasher;
private readonly IActivityRegistry _activityRegistry;
private readonly IList<ActivityNode> _nodes;
private readonly IList<ActivityCompletionCallbackEntry> _completionCallbackEntries = new List<ActivityCompletionCallbackEntry>();
@ -33,6 +34,7 @@ public class WorkflowExecutionContext : IExecutionContext
/// </summary>
public WorkflowExecutionContext(
IServiceProvider serviceProvider,
IHasher hasher,
string id,
string? correlationId,
Workflow workflow,
@ -47,6 +49,7 @@ public class WorkflowExecutionContext : IExecutionContext
CancellationToken cancellationToken)
{
_serviceProvider = serviceProvider;
_hasher = hasher;
_activityRegistry = activityRegistry;
Workflow = workflow;
Graph = graph;
@ -61,6 +64,7 @@ public class WorkflowExecutionContext : IExecutionContext
TriggerActivityId = triggerActivityId;
CancellationToken = cancellationToken;
NodeIdLookup = _nodes.ToDictionary(x => x.NodeId);
NodeHashLookup = _nodes.ToDictionary(x => Hash(x.NodeId));
NodeActivityLookup = _nodes.ToDictionary(x => x.Activity);
MemoryRegister = workflow.CreateRegister();
ExpressionExecutionContext = new ExpressionExecutionContext(serviceProvider, MemoryRegister, cancellationToken: cancellationToken);
@ -110,6 +114,11 @@ public class WorkflowExecutionContext : IExecutionContext
/// A map between activity IDs and <see cref="ActivityNode"/>s in the workflow graph.
/// </summary>
public IDictionary<string, ActivityNode> NodeIdLookup { get; }
/// <summary>
/// A map between hashed activity node IDs and <see cref="ActivityNode"/>s in the workflow graph.
/// </summary>
public IDictionary<string, ActivityNode> NodeHashLookup { get; }
/// <summary>
/// A map between <see cref="IActivity"/>s and <see cref="ActivityNode"/>s in the workflow graph.
@ -267,6 +276,13 @@ public class WorkflowExecutionContext : IExecutionContext
/// Returns the <see cref="ActivityNode"/> with the specified activity ID from the workflow graph.
/// </summary>
public ActivityNode FindNodeById(string nodeId) => NodeIdLookup[nodeId];
/// <summary>
/// Returns the <see cref="ActivityNode"/> with the specified hash of the activity node ID from the workflow graph.
/// </summary>
/// <param name="hash">The hash of the activity node ID.</param>
/// <returns>The <see cref="ActivityNode"/> with the specified hash of the activity node ID.</returns>
public ActivityNode FindNodeByHash(string hash) => NodeHashLookup[hash];
/// <summary>
/// Returns the <see cref="ActivityNode"/> containing the specified activity from the workflow graph.
@ -279,11 +295,16 @@ public class WorkflowExecutionContext : IExecutionContext
public IActivity FindActivityByNodeId(string nodeId) => FindNodeById(nodeId).Activity;
/// <summary>
///
/// Returns the <see cref="IActivity"/> with the specified ID from the workflow graph.
/// </summary>
/// <param name="activityId"></param>
/// <returns></returns>
public IActivity FindActivityByActivityId(string activityId) => FindNodeById(NodeIdLookup.Single(n => n.Key.Contains(activityId)).Value.NodeId).Activity;
/// <summary>
/// Returns the <see cref="IActivity"/> with the specified hash of the activity node ID from the workflow graph.
/// </summary>
/// <param name="hash">The hash of the activity node ID.</param>
/// <returns>The <see cref="IActivity"/> with the specified hash of the activity node ID.</returns>
public IActivity FindActivityByHash(string hash) => FindNodeByHash(hash).Activity;
/// <summary>
/// Returns a custom property with the specified key from the <see cref="Properties"/> dictionary.
@ -311,12 +332,12 @@ public class WorkflowExecutionContext : IExecutionContext
/// </summary>
public bool HasProperty(string name) => Properties.ContainsKey(name);
internal bool CanTransitionTo(WorkflowSubStatus targetSubStatus) => ValidateStatusTransition(targetSubStatus);
internal void TransitionTo(WorkflowSubStatus subStatus)
{
var targetStatus = GetMainStatus(subStatus);
if (!ValidateStatusTransition(SubStatus, subStatus))
throw new Exception($"Cannot transition from {Status} to {targetStatus}");
if (!ValidateStatusTransition(SubStatus))
throw new Exception($"Cannot transition from {SubStatus} to {subStatus}");
SubStatus = subStatus;
}
@ -390,11 +411,11 @@ public class WorkflowExecutionContext : IExecutionContext
_ => throw new ArgumentOutOfRangeException(nameof(subStatus), subStatus, null)
};
private bool ValidateStatusTransition(WorkflowSubStatus currentSubStatus, WorkflowSubStatus target)
private bool ValidateStatusTransition(WorkflowSubStatus targetSubStatus)
{
var currentMainStatus = GetMainStatus(currentSubStatus);
var currentMainStatus = GetMainStatus(SubStatus);
return currentMainStatus != WorkflowStatus.Finished;
}
private IEnumerable<MemoryRegister> GetMergedRegistersView() => new[] { MemoryRegister }.Concat(ActivityExecutionContexts.Select(x => x.ExpressionExecutionContext.Memory)).ToList();
private string Hash(string nodeId) => _hasher.Hash(nodeId);
}

6
src/modules/Elsa.Workflows.Core/Services/DefaultWorkflowExecutionContextFactory.cs
View file

@ -13,6 +13,7 @@ public class DefaultWorkflowExecutionContextFactory : IWorkflowExecutionContextF
private readonly IActivitySchedulerFactory _schedulerFactory;
private readonly IActivityRegistry _activityRegistry;
private readonly IWorkflowStateSerializer _workflowStateSerializer;
private readonly IHasher _hasher;
/// <summary>
/// Constructor.
@ -22,13 +23,15 @@ public class DefaultWorkflowExecutionContextFactory : IWorkflowExecutionContextF
IIdentityGraphService identityGraphService,
IActivitySchedulerFactory schedulerFactory,
IActivityRegistry activityRegistry,
IWorkflowStateSerializer workflowStateSerializer)
IWorkflowStateSerializer workflowStateSerializer,
IHasher hasher)
{
_activityVisitor = activityVisitor;
_identityGraphService = identityGraphService;
_schedulerFactory = schedulerFactory;
_activityRegistry = activityRegistry;
_workflowStateSerializer = workflowStateSerializer;
_hasher = hasher;
}
/// <inheritdoc />
@ -64,6 +67,7 @@ public class DefaultWorkflowExecutionContextFactory : IWorkflowExecutionContextF
// Setup a workflow execution context.
var workflowExecutionContext = new WorkflowExecutionContext(
serviceProvider,
_hasher,
instanceId,
correlationId,
workflow,

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

@ -10,7 +10,7 @@ public class Hasher : IHasher
/// <inheritdoc />
public string Hash(string value)
{
using var sha = HashAlgorithm.Create(HashAlgorithmName.SHA256.ToString())!;
using var sha = SHA256.Create();
return Hash(sha, value);
}

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

@ -114,7 +114,10 @@ public class WorkflowRunner : IWorkflowRunner
var triggerActivityId = options?.TriggerActivityId;
var workflowExecutionContext = await CreateWorkflowExecutionContextAsync(scope.ServiceProvider, workflow, workflowState.Id, correlationId, workflowState, input, default, triggerActivityId, cancellationToken);
var bookmarkId = options?.BookmarkId;
var nodeId = options?.ActivityNodeId;
var activityNodeId = options?.ActivityNodeId;
var activityId = options?.ActivityId;
var activityInstanceId = options?.ActivityInstanceId;
var activityHash = options?.ActivityHash;
if (bookmarkId != null)
{
@ -124,12 +127,30 @@ public class WorkflowRunner : IWorkflowRunner
if (bookmark != null)
workflowExecutionContext.ScheduleBookmark(bookmark);
}
else if (nodeId != null)
else if (activityNodeId != null)
{
// Schedule the activity.
var activity = workflowExecutionContext.FindActivityByNodeId(nodeId);
var activity = workflowExecutionContext.FindActivityByNodeId(activityNodeId);
workflowExecutionContext.ScheduleActivity(activity);
}
else if (activityHash != null)
{
// Schedule the activity.
var activity = workflowExecutionContext.FindActivityByHash(activityHash);
workflowExecutionContext.ScheduleActivity(activity);
}
else if (activityId != null)
{
// Schedule the activity.
var activity = workflowExecutionContext.FindActivityByActivityId(activityId);
workflowExecutionContext.ScheduleActivity(activity);
}
else if (activityInstanceId != null)
{
// Schedule the activity.
var activityExecutionContext = workflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => x.Id == activityInstanceId) ?? throw new Exception("No activity execution context found with the specified ID.");
workflowExecutionContext.ScheduleActivityExecutionContext(activityExecutionContext);
}
else
{
// Schedule the workflow itself.

3
src/modules/Elsa.Workflows.Runtime/Commands/DispatchWorkflowInstanceCommand.cs
View file

@ -8,5 +8,8 @@ public record DispatchWorkflowInstanceCommand(
string InstanceId,
string? BookmarkId = default,
string? ActivityId = default,
string? ActivityNodeId = default,
string? ActivityInstanceId = default,
string? ActivityHash = default,
IDictionary<string, object>? Input = default,
string? CorrelationId = default) : ICommand<Unit>;

17
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowHost.cs
View file

@ -13,22 +13,22 @@ public interface IWorkflowHost
/// The workflow definition.
/// </summary>
Workflow Workflow { get; set; }
/// <summary>
/// The workflow state.
/// </summary>
WorkflowState WorkflowState { get; set; }
/// <summary>
/// Returns a value indicating whether or not the specified workflow can start a new instance or not.
/// </summary>
Task<bool> CanStartWorkflowAsync(StartWorkflowHostOptions? options = default, CancellationToken cancellationToken = default);
/// <summary>
/// Start a new workflow instance and execute it.
/// </summary>
Task<StartWorkflowHostResult> StartWorkflowAsync(StartWorkflowHostOptions? options = default, CancellationToken cancellationToken = default);
/// <summary>
/// Resume an existing workflow instance.
/// </summary>
@ -37,7 +37,14 @@ public interface IWorkflowHost
public record StartWorkflowHostOptions(string? InstanceId = default, string? CorrelationId = default, IDictionary<string, object>? Input = default, string? TriggerActivityId = default);
public record ResumeWorkflowHostOptions(string? CorrelationId = default, string? BookmarkId = default, string? ActivityId = default, IDictionary<string, object>? Input = default);
public record ResumeWorkflowHostOptions(
string? CorrelationId = default,
string? BookmarkId = default,
string? ActivityId = default,
string? ActivityNodeId = default,
string? ActivityInstanceId = default,
string? ActivityHash = default,
IDictionary<string, object>? Input = default);
public record StartWorkflowHostResult(Diff<Bookmark> BookmarksDiff);

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

@ -92,7 +92,15 @@ public interface IWorkflowRuntime
public record StartWorkflowRuntimeOptions(string? CorrelationId = default, IDictionary<string, object>? Input = default, VersionOptions VersionOptions = default, string? TriggerActivityId = default, string? InstanceId = default);
public record ResumeWorkflowRuntimeOptions(string? CorrelationId = default, string? WorkflowInstanceId = default, string? BookmarkId = default, string? ActivityId = default, IDictionary<string, object>? Input = default);
public record ResumeWorkflowRuntimeOptions(
string? CorrelationId = default,
string? WorkflowInstanceId = default,
string? BookmarkId = default,
string? ActivityId = default,
string? ActivityNodeId = default,
string? ActivityInstanceId = default,
string? ActivityHash = default,
IDictionary<string, object>? Input = default);
public record CanStartWorkflowResult(string? InstanceId, bool CanStart);

11
src/modules/Elsa.Workflows.Runtime/Handlers/DispatchWorkflowRequestHandler.cs
View file

@ -38,7 +38,16 @@ internal class DispatchWorkflowRequestHandler :
public async Task<Unit> HandleAsync(DispatchWorkflowInstanceCommand command, CancellationToken cancellationToken)
{
var options = new ResumeWorkflowRuntimeOptions(command.CorrelationId, command.InstanceId, command.BookmarkId, command.ActivityId, command.Input);
var options = new ResumeWorkflowRuntimeOptions(
command.CorrelationId,
command.InstanceId,
command.BookmarkId,
command.ActivityId,
command.ActivityNodeId,
command.ActivityInstanceId,
command.ActivityHash,
command.Input);
await _workflowRuntime.ResumeWorkflowAsync(command.InstanceId, options, cancellationToken);
return Unit.Instance;

3
src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowInstanceRequest.cs
View file

@ -4,5 +4,8 @@ public record DispatchWorkflowInstanceRequest(
string InstanceId,
string? BookmarkId = default,
string? ActivityId = default,
string? ActivityNodeId = default,
string? ActivityInstanceId = default,
string? ActivityHash = default,
IDictionary<string, object>? Input = default,
string? CorrelationId = default);

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

@ -129,7 +129,15 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken);
var workflowHost = await _workflowHostFactory.CreateAsync(workflow, workflowState, cancellationToken);
var resumeWorkflowOptions = new ResumeWorkflowHostOptions(options.CorrelationId, options.BookmarkId, options.ActivityId, options.Input);
var resumeWorkflowOptions = new ResumeWorkflowHostOptions(
options.CorrelationId,
options.BookmarkId,
options.ActivityId,
options.ActivityNodeId,
options.ActivityInstanceId,
options.ActivityHash,
options.Input);
await workflowHost.ResumeWorkflowAsync(resumeWorkflowOptions, cancellationToken);
@ -296,7 +304,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
var hash = _hasher.Hash(workflowsFilter.ActivityTypeName, workflowsFilter.BookmarkPayload);
var correlationId = workflowsFilter.Options.CorrelationId;
var workflowInstanceId = workflowsFilter.Options.WorkflowInstanceId;
var filter = new BookmarkFilter { Hash = hash, CorrelationId = correlationId, WorkflowInstanceId = workflowInstanceId};
var filter = new BookmarkFilter { Hash = hash, CorrelationId = correlationId, WorkflowInstanceId = workflowInstanceId };
var bookmarks = await _bookmarkStore.FindManyAsync(filter, cancellationToken);
var collectedWorkflows = bookmarks.Select(b => new ResumableWorkflowMatch(b.WorkflowInstanceId, default, correlationId, b.BookmarkId)).ToList();
return collectedWorkflows;

11
src/modules/Elsa.Workflows.Runtime/Services/TaskBasedWorkflowDispatcher.cs
View file

@ -37,7 +37,16 @@ public class TaskBasedWorkflowDispatcher : IWorkflowDispatcher
/// <inheritdoc />
public async Task<DispatchWorkflowInstanceResponse> DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var command = new DispatchWorkflowInstanceCommand(request.InstanceId, request.BookmarkId, request.ActivityId, request.Input, request.CorrelationId);
var command = new DispatchWorkflowInstanceCommand(
request.InstanceId,
request.BookmarkId,
request.ActivityId,
request.ActivityNodeId,
request.ActivityInstanceId,
request.ActivityHash,
request.Input,
request.CorrelationId);
await _backgroundCommandSender.SendAsync(command, cancellationToken);
return new DispatchWorkflowInstanceResponse();
}

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

@ -38,7 +38,7 @@ public class WorkflowHost : IWorkflowHost
/// <inheritdoc />
public Workflow Workflow { get; set; }
/// <inheritdoc />
public WorkflowState WorkflowState { get; set; }
@ -84,7 +84,7 @@ public class WorkflowHost : IWorkflowHost
public async Task<ResumeWorkflowHostResult> ResumeWorkflowAsync(ResumeWorkflowHostOptions? options = default, CancellationToken cancellationToken = default)
{
var originalBookmarks = WorkflowState.Bookmarks.ToList();
if (WorkflowState.Status != WorkflowStatus.Running)
{
_logger.LogWarning("Attempt to resume workflow {WorkflowInstanceId} that is not in the Running state. The actual state is {ActualWorkflowStatus}", WorkflowState.Id, WorkflowState.Status);
@ -93,11 +93,22 @@ public class WorkflowHost : IWorkflowHost
var instanceId = WorkflowState.Id;
var input = options?.Input;
var runOptions = new RunWorkflowOptions(instanceId, options?.CorrelationId, options?.BookmarkId, options?.ActivityId, input);
var runOptions = new RunWorkflowOptions(
instanceId,
options?.CorrelationId,
options?.BookmarkId,
options?.ActivityId,
options?.ActivityNodeId,
options?.ActivityInstanceId,
options?.ActivityHash,
input
);
var workflowResult = await _workflowRunner.RunAsync(Workflow, WorkflowState, runOptions, cancellationToken);
WorkflowState = workflowResult.WorkflowState;
var updatedBookmarks = WorkflowState.Bookmarks;
return new ResumeWorkflowHostResult(Diff.For(originalBookmarks, updatedBookmarks));
}