diff --git a/.gitignore b/.gitignore index 8a28551f4..94ac694b2 100644 --- a/.gitignore +++ b/.gitignore @@ -78,5 +78,5 @@ unlist.sh /artifacts /docker/data/ - +docker/azurite-data docker/docker-compose-datadog.yml diff --git a/Directory.Packages.props b/Directory.Packages.props index ea4ed3123..a3a75fa44 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -31,10 +31,10 @@ - - - - + + + + @@ -123,37 +123,37 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -161,9 +161,9 @@ - - - - + + + + \ No newline at end of file diff --git a/Elsa.sln b/Elsa.sln index 6e9135689..b35fe3352 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -374,6 +374,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "adr", "adr", "{0A04B1FD-06C doc\adr\0001-record-architecture-decisions.md = doc\adr\0001-record-architecture-decisions.md doc\adr\0002-fault-propagation-from-child-to-parent-activities.md = doc\adr\0002-fault-propagation-from-child-to-parent-activities.md doc\adr\0003-direct-bookmark-management-in-workflowexecutioncontext.md = doc\adr\0003-direct-bookmark-management-in-workflowexecutioncontext.md + doc\adr\0004-token-centric-flowchart-execution-model.md = doc\adr\0004-token-centric-flowchart-execution-model.md doc\adr\graph.dot = doc\adr\graph.dot doc\adr\toc.md = doc\adr\toc.md EndProjectSection diff --git a/doc/adr/0004-token-centric-flowchart-execution-model.md b/doc/adr/0004-token-centric-flowchart-execution-model.md new file mode 100644 index 000000000..4c01e6536 --- /dev/null +++ b/doc/adr/0004-token-centric-flowchart-execution-model.md @@ -0,0 +1,83 @@ +# 4. Token-Centric Flowchart Execution Model + +Date: 2025-05-06 + +## Status + +Accepted + +## Context + +Elsa Workflows’ original flowchart used execution-count heuristics to drive joins, which fails in loops, XOR splits and resumable activities: + +- Loop-back edges never emit a “forward” token, stalling AND-joins. +- Counting executions across iterations causes premature or missed firings. +- Resumable activities (e.g. `Delay`) clear join state on resume. +- Users cannot declaratively control join semantics without deep framework hacks. + +We need a model that: + +1. Handles loops, forks, XORs and resumable activities reliably. +2. Lets designers choose per-activity join behavior. +3. Cleans up state to avoid memory leaks. +4. Supports cancellation of in-flight branches. + +## Decision + +Adopt a **token-centric** execution model with explicit **MergeMode** and **blocking**: + +1. **Tokens** + - On each activity completion, for each active outbound connection, emit a `Token` with: + - `FromActivityId`, `Outcome`, `ToActivityId`, + - Flags: `Consumed = false`, `Blocked = false`. + - Persist the list in `ActivityExecutionContext.Properties["Flowchart.Tokens"]`. + +2. **MergeMode** + - Query each target activity’s `MergeMode` via `GetMergeModeAsync(...)`. Supported values: + - **Race**: “first wins” + - **Stream**: “first wins, but don’t cancel ancestors” + - **Converge** (default): “wait for all” + - **Race** + 1. Cancel inbound ancestors (`CancelInboundAncestorsAsync`). + 2. If no existing blocked token for this inbound connection, schedule the target and then block all other inbound branches by emitting `Token.Block()` for each. + 3. Subsequent branches see their blocked token and simply consume it. + - **Stream** + - Same as Race except you do _not_ cancel inbound ancestors. + - **Converge** + - Wait until _every_ inbound connection for the target has at least one unblocked, unconsumed token. Then schedule once. + +3. **Scheduling Loop** + On each child completion: + - _Emit_ tokens for its outbound edges. + - _Consume_ any tokens whose `ToActivityId` matches the completed activity. + - _For each_ active outbound connection, inspect its target’s `MergeMode` and apply the rules above to decide whether to schedule it. + +4. **State Cleanup** + - After scheduling (or skipping) a target, remove any _consumed_ tokens whose `ToActivityId` equals the completed activity. + - When the flow has no pending work (`HasPendingWork()` is false), clear the entire token list and complete the flowchart. + - On activity cancellation (`OnTokenFlowActivityCanceledAsync`), remove _all_ tokens from or to that activity, then re-check for completion. + +## Sequence Diagram + +```mermaid +sequenceDiagram + participant A as Activity A + participant F as Flowchart + participant B as Activity B + + A-->>F: Completed(outcome="Done") + F->>F: Emit Token(A→B,Block=false,Consumed=false) + F->>F: Consume any inbound tokens for A + F->>F: Get B.MergeMode() + alt Race & first branch + F->>F: CancelInboundAncestors(B) + F-->>B: Schedule B + F->>F: Emit blocked Tokens for other inbound edges into B + else Race & later branch + F->>F: Consume blocked Token + else Converge until all arrived + Note over F: wait + end + F->>F: Purge consumed tokens for A + F-->>F: Complete if no pending work +``` \ No newline at end of file diff --git a/src/clients/Elsa.Api.Client/Extensions/ActivityExtensions.cs b/src/clients/Elsa.Api.Client/Extensions/ActivityExtensions.cs index 4990ea81a..251022162 100644 --- a/src/clients/Elsa.Api.Client/Extensions/ActivityExtensions.cs +++ b/src/clients/Elsa.Api.Client/Extensions/ActivityExtensions.cs @@ -1,5 +1,6 @@ using System.Text.Json.Nodes; using Elsa.Api.Client.Resources.WorkflowDefinitions.Models; +using Elsa.Api.Client.Shared.Enums; using Elsa.Api.Client.Shared.Models; namespace Elsa.Api.Client.Extensions; @@ -116,11 +117,21 @@ public static class ActivityExtensions /// Sets a value indicating whether the specified activity can trigger the workflow. /// public static void SetCanStartWorkflow(this JsonObject activity, bool value) => activity.SetProperty(JsonValue.Create(value), "customProperties", "canStartWorkflow"); + + public static MergeMode GetMergeMode(this JsonObject activity) + { + return activity.GetProperty("customProperties", "mergeMode") ?? MergeMode.Converge; + } + + public static void SetMergeMode(this JsonObject activity, MergeMode? value) + { + activity.SetProperty(JsonValue.Create(value), "customProperties", "mergeMode"); + } /// /// Gets the activities in the specified flowchart. /// - public static IEnumerable GetActivities(this JsonObject flowchart) => flowchart.GetProperty("activities")?.AsArray().AsEnumerable().Cast() ?? Array.Empty(); + public static IEnumerable GetActivities(this JsonObject flowchart) => flowchart.GetProperty("activities")?.AsArray().AsEnumerable().Cast() ?? []; /// /// Sets the activities in the specified flowchart. diff --git a/src/clients/Elsa.Api.Client/Extensions/JsonObjectExtensions.cs b/src/clients/Elsa.Api.Client/Extensions/JsonObjectExtensions.cs index b37f0f742..a66ea5a31 100644 --- a/src/clients/Elsa.Api.Client/Extensions/JsonObjectExtensions.cs +++ b/src/clients/Elsa.Api.Client/Extensions/JsonObjectExtensions.cs @@ -1,5 +1,7 @@ using System.Text.Json; using System.Text.Json.Nodes; +using System.Text.Json.Serialization; +using Elsa.Extensions; namespace Elsa.Api.Client.Extensions; @@ -24,7 +26,11 @@ public static class JsonObjectExtensions /// A representing the specified value. public static JsonNode SerializeToNode(this object value, JsonSerializerOptions? options = null) { - options ??= new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + options ??= (new JsonSerializerOptions + { + PropertyNamingPolicy = JsonNamingPolicy.CamelCase, + + }).WithConverters(new JsonStringEnumConverter()); return JsonSerializer.SerializeToNode(value, options)!; } diff --git a/src/clients/Elsa.Api.Client/Shared/Enums/MergeMode.cs b/src/clients/Elsa.Api.Client/Shared/Enums/MergeMode.cs new file mode 100644 index 000000000..d2f846440 --- /dev/null +++ b/src/clients/Elsa.Api.Client/Shared/Enums/MergeMode.cs @@ -0,0 +1,22 @@ +namespace Elsa.Api.Client.Shared.Enums; + +/// +/// Specifies the strategy for handling multiple inbound execution paths in a workflow. +/// +public enum MergeMode +{ + /// + /// Wait for all inbound paths before proceeding. + /// + Converge, + + /// + /// Proceed when any one inbound path completes; cancel all others. + /// + Race, + + /// + /// Proceed when any one inbound path completes; do not cancel others. + /// + Stream +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Container.cs b/src/modules/Elsa.Workflows.Core/Activities/Container.cs index 66a4f3dd3..cfe50e3b2 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Container.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Container.cs @@ -9,7 +9,7 @@ namespace Elsa.Workflows.Activities; public abstract class Container : Activity, IVariableContainer { /// - protected Container(string? source = default, int? line = default) : base(source, line) + protected Container(string? source = null, int? line = null) : base(source, line) { } diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs index f5d02dda9..a3dacfb73 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs @@ -1,5 +1,4 @@ using System.Runtime.CompilerServices; -using Elsa.Extensions; using Elsa.Workflows.Activities.Flowchart.Contracts; using Elsa.Workflows.Activities.Flowchart.Extensions; using Elsa.Workflows.Activities.Flowchart.Models; @@ -12,13 +11,16 @@ namespace Elsa.Workflows.Activities.Flowchart.Activities; /// /// Merge multiple branches into a single branch of execution. +/// Note that this activity is no longer necessary for either AND or OR merges, because all activities inherit the Join Kind property. +/// Use this activity if an explicit join step is desired. /// -[Activity("Elsa", "Branching", "Merge multiple branches into a single branch of execution.", DisplayName = "Join")] -[PublicAPI] +[Activity("Elsa", "Branching", "[Obsolete] - Explicitly merge multiple branches into a single branch of execution.", DisplayName = "Join")] +[UsedImplicitly] +[Obsolete("Each activity now supports the MergeMode property, making the use of this activity obsolete.", false)] public class FlowJoin : Activity, IJoinNode { /// - public FlowJoin([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + public FlowJoin([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line) { } @@ -35,15 +37,22 @@ public class FlowJoin : Activity, IJoinNode /// protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { - Flowchart.CancelAncestorActivatesAsync(context); + if(!Flowchart.UseTokenFlow) + await context.ParentActivityExecutionContext.CancelInboundAncestorsAsync(this); + await context.CompleteActivityAsync(); } protected override bool CanExecute(ActivityExecutionContext context) - => context.Get(Mode) switch + { + if(Flowchart.UseTokenFlow) + return true; + + return context.Get(Mode) switch { FlowJoinMode.WaitAny => true, FlowJoinMode.WaitAll => Flowchart.CanWaitAllProceed(context), _ => true }; + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowSwitch.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowSwitch.cs index 54c4e2dcb..e2be725af 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowSwitch.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowSwitch.cs @@ -19,7 +19,7 @@ namespace Elsa.Workflows.Activities.Flowchart.Activities; public class FlowSwitch : Activity { /// - public FlowSwitch([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + public FlowSwitch([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line) { } diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Counters.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Counters.cs new file mode 100644 index 000000000..03db3b0d1 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Counters.cs @@ -0,0 +1,260 @@ +using Elsa.Extensions; +using Elsa.Workflows.Activities.Flowchart.Contracts; +using Elsa.Workflows.Activities.Flowchart.Extensions; +using Elsa.Workflows.Activities.Flowchart.Models; +using Elsa.Workflows.Options; +using Elsa.Workflows.Signals; + +namespace Elsa.Workflows.Activities.Flowchart.Activities; + +public partial class Flowchart +{ + private const string ScopeProperty = "FlowScope"; + private const string BackwardConnectionActivityInput = "BackwardConnection"; + + private async ValueTask OnChildCompletedCounterBasedLogicAsync(ActivityCompletedContext context) + { + var flowchartContext = context.TargetContext; + var completedActivityContext = context.ChildContext; + var completedActivity = completedActivityContext.Activity; + var result = context.Result; + + if (flowchartContext.Activity != this) + { + throw new Exception("Target context activity must be this flowchart"); + } + + // If the completed activity's status is anything but "Completed", do not schedule its outbound activities. + if (completedActivityContext.Status != ActivityStatus.Completed) + { + return; + } + + // If the complete activity is a terminal node, complete the flowchart immediately. + if (completedActivity is ITerminalNode) + { + await flowchartContext.CompleteActivityAsync(); + return; + } + + // Determine the outcomes from the completed activity + var outcomes = result is Outcomes o ? o : Outcomes.Default; + + // Schedule the outbound activities + var flowGraph = flowchartContext.GetFlowGraph(); + var flowScope = GetFlowScope(flowchartContext); + var completedActivityExecutedByBackwardConnection = completedActivityContext.ActivityInput.GetValueOrDefault(BackwardConnectionActivityInput); + bool hasScheduledActivity = await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, completedActivity, outcomes, completedActivityExecutedByBackwardConnection); + + // If there are not any outbound connections, complete the flowchart activity if there is no other pending work + if (!hasScheduledActivity) + { + await CompleteIfNoPendingWorkAsync(flowchartContext); + } + } + + private FlowScope GetFlowScope(ActivityExecutionContext context) + { + return context.GetProperty(ScopeProperty, () => new FlowScope()); + } + + /// + /// Schedules outbound activities based on the flowchart's structure and execution state. + /// This method determines whether an activity should be scheduled based on visited connections, + /// forward traversal rules, and backward connections. + /// + /// The flowchart containing the activities. + /// The graph representation of the flowchart. + /// Tracks activity and connection visits. + /// The execution context of the flowchart. + /// The current activity being processed. + /// The outcomes that determine which connections were followed. + /// Indicates if the completed activity was executed due to a backward connection. + /// True if at least one activity was scheduled; otherwise, false. + private async ValueTask ScheduleOutboundActivitiesAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, IActivity activity, Outcomes outcomes, bool completedActivityExecutedByBackwardConnection = false) + { + var hasScheduledActivity = false; + + // Check if the activity is dangling (i.e., it is not reachable from the flowchart graph) + if (flowGraph.IsDanglingActivity(activity)) + { + throw new Exception($"Activity {activity.Id} is not reachable from the flowchart graph. Unable to schedule it's outbound activities."); + } + + // Register the activity as visited unless it was executed due to a backward connection + if (!completedActivityExecutedByBackwardConnection) + { + flowScope.RegisterActivityVisit(activity); + } + + // Process each outbound connection from the current activity + foreach (var outboundConnection in flowGraph.GetOutboundConnections(activity)) + { + var connectionFollowed = outcomes.Names.Contains(outboundConnection.Source.Port); + flowScope.RegisterConnectionVisit(outboundConnection, connectionFollowed); + var outboundActivity = outboundConnection.Target.Activity; + + // Determine scheduling strategy based on connection type + if (flowGraph.IsBackwardConnection(outboundConnection, out var backwardConnectionIsValid)) + { + hasScheduledActivity |= await ScheduleBackwardConnectionActivityAsync(flowGraph, flowchartContext, outboundConnection, outboundActivity, connectionFollowed, backwardConnectionIsValid); + } + else if (outboundActivity is not IJoinNode) + { + hasScheduledActivity |= await ScheduleNonJoinActivityAsync(flowGraph, flowScope, flowchartContext, outboundActivity); + } + else + { + hasScheduledActivity |= await ScheduleJoinActivityAsync(flowGraph, flowScope, flowchartContext, outboundConnection, outboundActivity); + } + } + + return hasScheduledActivity; + } + + /// + /// Schedules an outbound activity that originates from a backward connection. + /// + private async ValueTask ScheduleBackwardConnectionActivityAsync(FlowGraph flowGraph, ActivityExecutionContext flowchartContext, Connection outboundConnection, IActivity outboundActivity, bool connectionFollowed, bool backwardConnectionIsValid) + { + if (!connectionFollowed) + { + return false; + } + + if (!backwardConnectionIsValid) + { + throw new Exception($"Invalid backward connection: Every path from the source ('{outboundConnection.Source.Activity.Id}') must go through the target ('{outboundConnection.Target.Activity.Id}') when tracing back to the start."); + } + + var scheduleWorkOptions = new ScheduleWorkOptions + { + CompletionCallback = OnChildCompletedCounterBasedLogicAsync, + Input = new Dictionary() + { + { + BackwardConnectionActivityInput, true + } + } + }; + + await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions); + return true; + } + + /// + /// Schedules a non-join activity if all its forward inbound connections have been visited. + /// + private async ValueTask ScheduleNonJoinActivityAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, IActivity outboundActivity) + { + if (!flowScope.AllInboundConnectionsVisited(flowGraph, outboundActivity)) + { + return false; + } + + if (flowScope.HasFollowedInboundConnection(flowGraph, outboundActivity)) + { + await flowchartContext.ScheduleActivityAsync(outboundActivity, OnChildCompletedCounterBasedLogicAsync); + return true; + } + else + { + // Propagate skipped connections by scheduling with Outcomes.Empty + return await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, outboundActivity, Outcomes.Empty); + } + } + + /// + /// Schedules a join activity based on inbound connection statuses. + /// + private async ValueTask ScheduleJoinActivityAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, Connection outboundConnection, IActivity outboundActivity) + { + // Ignore the connection if the join activity has already completed (JoinAny scenario) + if (flowScope.ShouldIgnoreConnection(outboundConnection, outboundActivity)) + { + return false; + } + + // Schedule the join activity only if at least one inbound connection was followed + if (!flowScope.HasFollowedInboundConnection(flowGraph, outboundActivity)) + { + if (flowScope.AllInboundConnectionsVisited(flowGraph, outboundActivity)) + { + // Propagate skipped connections by scheduling with Outcomes.Empty + return await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, outboundActivity, Outcomes.Empty); + } + + return false; + } + + // Check for an existing execution context for the join activity + var joinContext = flowchartContext.WorkflowExecutionContext.ActivityExecutionContexts.LastOrDefault(x => + x.ParentActivityExecutionContext == flowchartContext && + x.Activity == outboundActivity && + x.Status is ActivityStatus.Pending or ActivityStatus.Running); + + // If the join activity was already scheduled, do not schedule it again + if (joinContext == null) + { + var activityScheduled = flowchartContext.WorkflowExecutionContext.Scheduler.List().Any(workItem => workItem.Owner == flowchartContext && workItem.Activity == outboundActivity); + if (activityScheduled) + { + return true; + } + } + + if (joinContext is not { Status: ActivityStatus.Running }) + { + var scheduleWorkOptions = new ScheduleWorkOptions + { + CompletionCallback = OnChildCompletedCounterBasedLogicAsync, + ExistingActivityExecutionContext = joinContext + }; + await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions); + return true; + } + else + { + return false; + } + } + + public static bool CanWaitAllProceed(ActivityExecutionContext context) + { + var flowchartContext = context.ParentActivityExecutionContext!; + var flowchart = (Flowchart)flowchartContext.Activity; + var flowGraph = flowchartContext.GetFlowGraph(); + var flowScope = flowchart.GetFlowScope(flowchartContext); + var activity = context.Activity; + + return flowScope.AllInboundConnectionsVisited(flowGraph, activity); + } + + private async ValueTask OnScheduleOutcomesAsync(ScheduleActivityOutcomes signal, SignalContext context) + { + var flowchartContext = context.ReceiverActivityExecutionContext; + var schedulingActivityContext = context.SenderActivityExecutionContext; + var schedulingActivity = schedulingActivityContext.Activity; + var outcomes = signal.Outcomes; + var outboundConnections = Connections.Where(connection => connection.Source.Activity == schedulingActivity && outcomes.Contains(connection.Source.Port!)).ToList(); + var outboundActivities = outboundConnections.Select(x => x.Target.Activity).ToList(); + + if (outboundActivities.Any()) + { + foreach (var activity in outboundActivities) + await flowchartContext.ScheduleActivityAsync(activity, OnChildCompletedCounterBasedLogicAsync); + } + } + + private async ValueTask OnCounterFlowActivityCanceledAsync(CancelSignal signal, SignalContext context) + { + var flowchartContext = context.ReceiverActivityExecutionContext; + await CompleteIfNoPendingWorkAsync(flowchartContext); + var flowchart = (Flowchart)flowchartContext.Activity; + var flowGraph = flowchartContext.GetFlowGraph(); + var flowScope = flowchart.GetFlowScope(flowchartContext); + + // Propagate canceled connections visited count by scheduling with Outcomes.Empty + await flowchart.ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, context.SenderActivityExecutionContext.Activity, Outcomes.Empty); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Tokens.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Tokens.cs new file mode 100644 index 000000000..bfba1dda7 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Tokens.cs @@ -0,0 +1,114 @@ +using Elsa.Extensions; +using Elsa.Workflows.Activities.Flowchart.Extensions; +using Elsa.Workflows.Activities.Flowchart.Models; +using Elsa.Workflows.Signals; + +namespace Elsa.Workflows.Activities.Flowchart.Activities; + +public partial class Flowchart +{ + private const string TokenStoreKey = "Flowchart.Tokens"; + + private async ValueTask OnChildCompletedTokenBasedLogicAsync(ActivityCompletedContext ctx) + { + var flowContext = ctx.TargetContext; + var completedActivity = ctx.ChildContext.Activity; + var flowGraph = flowContext.GetFlowGraph(); + + // Emit tokens. + var outcomes = (ctx.Result as Outcomes ?? Outcomes.Default).Names; + var outboundConnections = flowGraph.GetOutboundConnections(completedActivity); + var activeOutboundConnections = outboundConnections.Where(x => outcomes.Contains(x.Source.Port)).Distinct().ToList(); + var tokens = GetTokenList(flowContext); + + foreach (var connection in activeOutboundConnections) + tokens.Add(Token.Create(connection.Source.Activity, connection.Target.Activity, connection.Source.Port)); + + // Consume tokens. + var inboundTokens = tokens.Where(t => t.ToActivityId == completedActivity.Id && t is { Consumed: false, Blocked: false }).ToList(); + foreach (var t in inboundTokens) + t.Consume(); + + // Schedule next activities. + foreach (var connection in activeOutboundConnections) + { + var targetActivity = connection.Target.Activity; + var mergeMode = await targetActivity.GetMergeModeAsync(ctx.ChildContext); + + if (mergeMode is MergeMode.Stream or MergeMode.Race) + { + if (mergeMode == MergeMode.Race) + await flowContext.CancelInboundAncestorsAsync(targetActivity); + + // Check if there is any blocking token preventing the activity from being scheduled. + var existingBlockedToken = tokens.FirstOrDefault(t => t.ToActivityId == targetActivity.Id && t.FromActivityId == connection.Source.Activity.Id && t.Outcome == connection.Source.Port && t.Blocked); + + if (existingBlockedToken == null) + { + // Schedule the target activity. + await flowContext.ScheduleActivityAsync(targetActivity, OnChildCompletedTokenBasedLogicAsync); + + // And block other inbound connections. + var otherInboundConnections = flowGraph.GetForwardInboundConnections(targetActivity).Where(x => x.Source.Activity != completedActivity).ToList(); + + foreach (var inboundConnection in otherInboundConnections) + { + var blockedToken = Token.Create(inboundConnection.Source.Activity, inboundConnection.Target.Activity, inboundConnection.Source.Port).Block(); + tokens.Add(blockedToken); + } + } + else + { + // Consume the block. + existingBlockedToken.Consume(); + } + } + else + { + // Wait for all inbound tokens to be consumed before scheduling the target activity. + var inboundConnections = flowGraph.GetForwardInboundConnections(targetActivity); + var hasUnconsumed = inboundConnections.Any(inbound => + tokens.Any(t => t is { Consumed: false, Blocked: false } && t.ToActivityId == inbound.Source.Activity.Id) + ); + + if (!hasUnconsumed) + { + await flowContext.ScheduleActivityAsync(targetActivity, OnChildCompletedTokenBasedLogicAsync); + } + } + } + + // Complete flow if done. + var hasPendingWork = flowContext.HasPendingWork(); + + if (!hasPendingWork) + { + tokens.Clear(); + await flowContext.CompleteActivityAsync(); + } + + // Purge tokens. + tokens.RemoveWhere(t => t.ToActivityId == completedActivity.Id && t.Consumed); + } + + private async ValueTask OnTokenFlowActivityCanceledAsync(CancelSignal signal, SignalContext context) + { + var flowchartContext = context.ReceiverActivityExecutionContext; + var cancelledActivityContext = context.SenderActivityExecutionContext; + + // Remove all tokens from and to this activity. + var tokenList = GetTokenList(flowchartContext); + tokenList.RemoveWhere(x => x.FromActivityId == cancelledActivityContext.Activity.Id || x.ToActivityId == cancelledActivityContext.Activity.Id); + await CompleteIfNoPendingWorkAsync(flowchartContext); + } + + internal List GetTokenList(ActivityExecutionContext context) + { + if (context.Properties.TryGetValue(TokenStoreKey, out var obj) && obj is List list) + return list; + + var newList = new List(); + context.Properties[TokenStoreKey] = newList; + return newList; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs index 65df2d963..e5b868ab4 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs @@ -1,10 +1,8 @@ using System.ComponentModel; using System.Runtime.CompilerServices; -using Elsa.Extensions; -using Elsa.Workflows.Activities.Flowchart.Contracts; +using Elsa.Workflows.Activities.Flowchart.Extensions; using Elsa.Workflows.Activities.Flowchart.Models; using Elsa.Workflows.Attributes; -using Elsa.Workflows.Options; using Elsa.Workflows.Signals; namespace Elsa.Workflows.Activities.Flowchart.Activities; @@ -14,11 +12,12 @@ namespace Elsa.Workflows.Activities.Flowchart.Activities; /// [Activity("Elsa", "Flow", "A flowchart is a collection of activities and connections between them.")] [Browsable(false)] -public class Flowchart : Container +public partial class Flowchart : Container { - private const string ScopeProperty = "FlowScope"; - private const string GraphTransientProperty = "FlowGraph"; - private const string BackwardConnectionActivityInput = "BackwardConnection"; + /// + /// Set this to false from your program file in case you wish to use the old counter based model. + /// + public static bool UseTokenFlow = true; /// public Flowchart([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line) @@ -41,7 +40,7 @@ public class Flowchart : Container /// protected override async ValueTask ScheduleChildrenAsync(ActivityExecutionContext context) { - var startActivity = GetStartActivity(context); + var startActivity = this.GetStartActivity(context.WorkflowExecutionContext.TriggerActivityId); if (startActivity == null) { @@ -50,350 +49,9 @@ public class Flowchart : Container return; } - // Schedule the start activity. await context.ScheduleActivityAsync(startActivity, OnChildCompletedAsync); } - private IActivity? GetStartActivity(ActivityExecutionContext context) - { - // If there's a trigger that triggered this workflow, use that. - var triggerActivityId = context.WorkflowExecutionContext.TriggerActivityId; - var triggerActivity = triggerActivityId != null ? Activities.FirstOrDefault(x => x.Id == triggerActivityId) : null; - - if (triggerActivity != null) - return triggerActivity; - - // If an explicit Start activity was provided, use that. - if (Start != null) - return Start; - - // If there is a Start activity on the flowchart, use that. - var startActivity = Activities.FirstOrDefault(x => x is Start); - - if (startActivity != null) - return startActivity; - - // If there's an activity marked as "Can Start Workflow", use that. - var canStartWorkflowActivity = Activities.FirstOrDefault(x => x.GetCanStartWorkflow()); - - if (canStartWorkflowActivity != null) - return canStartWorkflowActivity; - - // If there is a single activity that has no inbound connections, use that. - var root = GetRootActivity(); - - if (root != null) - return root; - - // If no start activity found, return the first activity. - return Activities.FirstOrDefault(); - } - - /// - /// Checks if there is any pending work for the flowchart. - /// - private bool HasPendingWork(ActivityExecutionContext context) - { - var workflowExecutionContext = context.WorkflowExecutionContext; - var activityIds = Activities.Select(x => x.Id).ToList(); - var children = context.Children; - var hasRunningActivityInstances = children.Where(x => activityIds.Contains(x.Activity.Id)).Any(x => x.Status == ActivityStatus.Running); - - var hasPendingWork = workflowExecutionContext.Scheduler.List().Any(workItem => - { - var ownerInstanceId = workItem.Owner?.Id; - - if (ownerInstanceId == null) - return false; - - if (ownerInstanceId == context.Id) - return true; - - var ownerContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.Id == ownerInstanceId); - var ancestors = ownerContext.GetAncestors().ToList(); - - return ancestors.Any(x => x == context); - }); - - return hasRunningActivityInstances || hasPendingWork; - } - - private IActivity? GetRootActivity() - { - // Get the first activity that has no inbound connections. - var query = - from activity in Activities - let inboundConnections = Connections.Any(x => x.Target.Activity == activity) - where !inboundConnections - select activity; - - var rootActivity = query.FirstOrDefault(); - return rootActivity; - } - - private FlowGraph GetFlowGraph(ActivityExecutionContext context) - { - // Store in TransientProperties so FlowChart is not persisted in WorkflowState - return context.TransientProperties.GetOrAdd(GraphTransientProperty, () => new FlowGraph(Connections, GetStartActivity(context))); - } - - private FlowScope GetFlowScope(ActivityExecutionContext context) - { - return context.GetProperty(ScopeProperty, () => new FlowScope()); - } - - private async ValueTask OnChildCompletedAsync(ActivityCompletedContext context) - { - var flowchartContext = context.TargetContext; - var completedActivityContext = context.ChildContext; - var completedActivity = completedActivityContext.Activity; - var result = context.Result; - - if (flowchartContext.Activity != this) - { - throw new Exception("Target context activity must be this flowchart"); - } - - // If the completed activity's status is anything but "Completed", do not schedule its outbound activities. - if (completedActivityContext.Status != ActivityStatus.Completed) - { - return; - } - - // If the complete activity is a terminal node, complete the flowchart immediately. - if (completedActivity is ITerminalNode) - { - await flowchartContext.CompleteActivityAsync(); - return; - } - - // Determine the outcomes from the completed activity - var outcomes = result is Outcomes o ? o : Outcomes.Default; - - // Schedule the outbound activities - var flowGraph = GetFlowGraph(flowchartContext); - var flowScope = GetFlowScope(flowchartContext); - var completedActivityExcecutedByBackwardConnection = completedActivityContext.ActivityInput.GetValueOrDefault(BackwardConnectionActivityInput); - bool hasScheduledActivity = await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, completedActivity, outcomes, completedActivityExcecutedByBackwardConnection); - - // If there are not any outbound connections, complete the flowchart activity if there is no other pending work - if (!hasScheduledActivity) - { - await CompleteIfNoPendingWorkAsync(flowchartContext); - } - } - - /// - /// Schedules outbound activities based on the flowchart's structure and execution state. - /// This method determines whether an activity should be scheduled based on visited connections, - /// forward traversal rules, and backward connections. - /// - /// The flowchart containing the activities. - /// The graph representation of the flowchart. - /// Tracks activity and connection visits. - /// The execution context of the flowchart. - /// The current activity being processed. - /// The outcomes that determine which connections were followed. - /// Indicates if the completed activity was executed due to a backward connection. - /// True if at least one activity was scheduled; otherwise, false. - private async ValueTask ScheduleOutboundActivitiesAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, IActivity activity, Outcomes outcomes, bool completedActivityExecutedByBackwardConnection = false) - { - bool hasScheduledActivity = false; - - // Check if the activity is dangling (i.e., it is not reachable from the flowchart graph) - if (flowGraph.IsDanglingActivity(activity)) - { - throw new Exception($"Activity {activity.Id} is not reachable from the flowchart graph. Unable to schedule it's outbound activities."); - } - - // Register the activity as visited unless it was executed due to a backward connection - if (!completedActivityExecutedByBackwardConnection) - { - flowScope.RegisterActivityVisit(activity); - } - - // Process each outbound connection from the current activity - foreach (var outboundConnection in flowGraph.GetOutboundConnections(activity)) - { - bool connectionFollowed = outcomes.Names.Contains(outboundConnection.Source.Port); - flowScope.RegisterConnectionVisit(outboundConnection, connectionFollowed); - var outboundActivity = outboundConnection.Target.Activity; - - // Determine scheduling strategy based on connection type - if (flowGraph.IsBackwardConnection(outboundConnection, out bool backwardConnectionIsValid)) - { - hasScheduledActivity |= await ScheduleBackwardConnectionActivityAsync(flowGraph, flowchartContext, outboundConnection, outboundActivity, connectionFollowed, backwardConnectionIsValid); - } - else if (outboundActivity is not IJoinNode) - { - hasScheduledActivity |= await ScheduleNonJoinActivityAsync(flowGraph, flowScope, flowchartContext, outboundActivity); - } - else - { - hasScheduledActivity |= await ScheduleJoinActivityAsync(flowGraph, flowScope, flowchartContext, outboundConnection, outboundActivity); - } - } - return hasScheduledActivity; - } - - /// - /// Schedules an outbound activity that originates from a backward connection. - /// - private async ValueTask ScheduleBackwardConnectionActivityAsync(FlowGraph flowGraph, ActivityExecutionContext flowchartContext, Connection outboundConnection, IActivity outboundActivity, bool connectionFollowed, bool backwardConnectionIsValid) - { - if (!connectionFollowed) - { - return false; - } - - if (!backwardConnectionIsValid) - { - throw new Exception($"Invalid backward connection: Every path from the source ('{outboundConnection.Source.Activity.Id}') must go through the target ('{outboundConnection.Target.Activity.Id}') when tracing back to the start."); - } - - var scheduleWorkOptions = new ScheduleWorkOptions - { - CompletionCallback = OnChildCompletedAsync, - Input = new Dictionary() { { BackwardConnectionActivityInput, true } } - }; - - await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions); - return true; - } - - /// - /// Schedules a non-join activity if all its forward inbound connections have been visited. - /// - private async ValueTask ScheduleNonJoinActivityAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, IActivity outboundActivity) - { - if (!flowScope.AllInboundConnectionsVisited(flowGraph, outboundActivity)) - { - return false; - } - - if (flowScope.HasFollowedInboundConnection(flowGraph, outboundActivity)) - { - await flowchartContext.ScheduleActivityAsync(outboundActivity, OnChildCompletedAsync); - return true; - } - else - { - // Propagate skipped connections by scheduling with Outcomes.Empty - return await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, outboundActivity, Outcomes.Empty); - } - } - - /// - /// Schedules a join activity based on inbound connection statuses. - /// - private async ValueTask ScheduleJoinActivityAsync(FlowGraph flowGraph, FlowScope flowScope, ActivityExecutionContext flowchartContext, Connection outboundConnection, IActivity outboundActivity) - { - // Ignore the connection if the join activity has already completed (JoinAny scenario) - if (flowScope.ShouldIgnoreConnection(outboundConnection, outboundActivity)) - { - return false; - } - - // Schedule the join activity only if at least one inbound connection was followed - if (!flowScope.HasFollowedInboundConnection(flowGraph, outboundActivity)) - { - if (flowScope.AllInboundConnectionsVisited(flowGraph, outboundActivity)) - { - // Propagate skipped connections by scheduling with Outcomes.Empty - return await ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, outboundActivity, Outcomes.Empty); - } - return false; - } - - // Check for an existing execution context for the join activity - var joinContext = flowchartContext.WorkflowExecutionContext.ActivityExecutionContexts.LastOrDefault(x => - x.ParentActivityExecutionContext == flowchartContext && - x.Activity == outboundActivity && - x.Status is ActivityStatus.Pending or ActivityStatus.Running); - - // If the join activity was already scheduled, do not schedule it again - if (joinContext == null) - { - var activityScheduled = flowchartContext.WorkflowExecutionContext.Scheduler.List().Any(workItem => workItem.Owner == flowchartContext && workItem.Activity == outboundActivity); - if (activityScheduled) - { - return true; - } - } - - if (joinContext is not { Status: ActivityStatus.Running }) - { - var scheduleWorkOptions = new ScheduleWorkOptions - { - CompletionCallback = OnChildCompletedAsync, - ExistingActivityExecutionContext = joinContext - }; - await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions); - return true; - } - else - { - return false; - } - } - - public static bool CanWaitAllProceed(ActivityExecutionContext context) - { - var flowchartContext = context.ParentActivityExecutionContext!; - var flowchart = (Flowchart)flowchartContext.Activity; - var flowGraph = flowchart.GetFlowGraph(flowchartContext); - var flowScope = flowchart.GetFlowScope(flowchartContext); - var activity = context.Activity; - - return flowScope.AllInboundConnectionsVisited(flowGraph, activity); - } - - public static async void CancelAncestorActivatesAsync(ActivityExecutionContext context) - { - var flowchartContext = context.ParentActivityExecutionContext!; - var flowchart = (Flowchart)flowchartContext.Activity; - var flowGraph = flowchart.GetFlowGraph(flowchartContext); - var ancestorActivities = flowGraph.GetAncestorActivities(context.Activity); - var inboundActivityExecutionContexts = context.WorkflowExecutionContext.ActivityExecutionContexts.Where(x => ancestorActivities.Contains(x.Activity) && x.ParentActivityExecutionContext == flowchartContext).ToList(); - - // Cancel each ancestor activity. - foreach (var activityExecutionContext in inboundActivityExecutionContexts) - { - await activityExecutionContext.CancelActivityAsync(); - } - } - - private async Task CompleteIfNoPendingWorkAsync(ActivityExecutionContext context) - { - var hasPendingWork = HasPendingWork(context); - - if (!hasPendingWork) - { - var hasFaultedActivities = context.Children.Any(x => x.Status == ActivityStatus.Faulted); - - if (!hasFaultedActivities) - { - await context.CompleteActivityAsync(); - } - } - } - - private async ValueTask OnScheduleOutcomesAsync(ScheduleActivityOutcomes signal, SignalContext context) - { - var flowchartContext = context.ReceiverActivityExecutionContext; - var schedulingActivityContext = context.SenderActivityExecutionContext; - var schedulingActivity = schedulingActivityContext.Activity; - var outcomes = signal.Outcomes; - var outboundConnections = Connections.Where(connection => connection.Source.Activity == schedulingActivity && outcomes.Contains(connection.Source.Port!)).ToList(); - var outboundActivities = outboundConnections.Select(x => x.Target.Activity).ToList(); - - if (outboundActivities.Any()) - { - // Schedule each child. - foreach (var activity in outboundActivities) await flowchartContext.ScheduleActivityAsync(activity, OnChildCompletedAsync); - } - } - private async ValueTask OnScheduleChildActivityAsync(ScheduleChildActivity signal, SignalContext context) { var flowchartContext = context.ReceiverActivityExecutionContext; @@ -402,7 +60,7 @@ public class Flowchart : Container if (activityExecutionContext != null) { - await flowchartContext.ScheduleActivityAsync(activityExecutionContext.Activity, new ScheduleWorkOptions + await flowchartContext.ScheduleActivityAsync(activityExecutionContext.Activity, new() { ExistingActivityExecutionContext = activityExecutionContext, CompletionCallback = OnChildCompletedAsync, @@ -411,7 +69,7 @@ public class Flowchart : Container } else { - await flowchartContext.ScheduleActivityAsync(activity, new ScheduleWorkOptions + await flowchartContext.ScheduleActivityAsync(activity, new() { CompletionCallback = OnChildCompletedAsync, Input = signal.Input @@ -419,16 +77,25 @@ public class Flowchart : Container } } - private async ValueTask OnActivityCanceledAsync(CancelSignal signal, SignalContext context) + private ValueTask OnChildCompletedAsync(ActivityCompletedContext context) { - await CompleteIfNoPendingWorkAsync(context.ReceiverActivityExecutionContext); + return UseTokenFlow + ? OnChildCompletedTokenBasedLogicAsync(context) + : OnChildCompletedCounterBasedLogicAsync(context); + } - var flowchartContext = context.ReceiverActivityExecutionContext!; - var flowchart = (Flowchart)flowchartContext.Activity; - var flowGraph = flowchart.GetFlowGraph(flowchartContext); - var flowScope = flowchart.GetFlowScope(flowchartContext); + private ValueTask OnActivityCanceledAsync(CancelSignal signal, SignalContext context) + { + return UseTokenFlow + ? OnTokenFlowActivityCanceledAsync(signal, context) + : OnCounterFlowActivityCanceledAsync(signal, context); + } - // Propagate canceled connections visited count by scheduling with Outcomes.Empty - await flowchart.ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, context.SenderActivityExecutionContext.Activity, Outcomes.Empty); + private async Task CompleteIfNoPendingWorkAsync(ActivityExecutionContext context) + { + var hasPendingWork = context.HasPendingWork(); + + if (!hasPendingWork) + await context.CompleteActivityAsync(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/ActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/ActivityExecutionContextExtensions.cs new file mode 100644 index 000000000..317a54abb --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/ActivityExecutionContextExtensions.cs @@ -0,0 +1,106 @@ +using Elsa.Extensions; +using Elsa.Workflows.Activities.Flowchart.Models; + +namespace Elsa.Workflows.Activities.Flowchart.Extensions; + +public static class ActivityExecutionContextExtensions +{ + private const string GraphTransientProperty = "FlowGraph"; + + public static IActivity? GetStartActivity(this Activities.Flowchart flowchart, string? triggerActivityId) + { + var activities = flowchart.Activities; + + // If there's a trigger that triggered this workflow, use that. + var triggerActivity = triggerActivityId != null ? activities.FirstOrDefault(x => x.Id == triggerActivityId) : null; + + if (triggerActivity != null) + return triggerActivity; + + // If an explicit Start activity was provided, use that. + if (flowchart.Start != null) + return flowchart.Start; + + // If there is a Start activity on the flowchart, use that. + var startActivity = activities.FirstOrDefault(x => x is Start); + + if (startActivity != null) + return startActivity; + + // If there's an activity marked as "Can Start Workflow", use that. + var canStartWorkflowActivity = activities.FirstOrDefault(x => x.GetCanStartWorkflow()); + + if (canStartWorkflowActivity != null) + return canStartWorkflowActivity; + + // If there is a single activity that has no inbound connections, use that. + var root = flowchart.GetRootActivity(); + + if (root != null) + return root; + + // If no start activity found, return the first activity. + return activities.FirstOrDefault(); + } + + /// + /// Checks if there is any pending work for the flowchart. + /// + internal static bool HasPendingWork(this ActivityExecutionContext context) + { + var flowchart = (Activities.Flowchart)context.Activity; + var workflowExecutionContext = context.WorkflowExecutionContext; + var activityIds = flowchart.Activities.Select(x => x.Id).ToList(); + var children = context.Children; + var hasRunningActivityInstances = children.Where(x => activityIds.Contains(x.Activity.Id)).Any(x => x.Status == ActivityStatus.Running); + var hasUnconsumedTokens = flowchart.GetTokenList(context).Any(x => x is { Consumed: false, Blocked: false }); + var hasFaulted = context.HasFaultedChildren(); + + var hasPendingWork = workflowExecutionContext.Scheduler.List().Any(workItem => + { + var ownerInstanceId = workItem.Owner?.Id; + + if (ownerInstanceId == null) + return false; + + if (ownerInstanceId == context.Id) + return true; + + var ownerContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.Id == ownerInstanceId); + var ancestors = ownerContext.GetAncestors().ToList(); + + return ancestors.Any(x => x == context); + }); + + return hasRunningActivityInstances || hasPendingWork || hasUnconsumedTokens || hasFaulted; + } + + internal static bool HasFaultedChildren(this ActivityExecutionContext context) + { + return context.Children.Any(x => x.Status == ActivityStatus.Faulted); + } + + internal static FlowGraph GetFlowGraph(this ActivityExecutionContext context) + { + // Store in TransientProperties so FlowChart is not persisted in WorkflowState + var flowchart = (Activities.Flowchart)context.Activity; + var startActivity = flowchart.GetStartActivity(context.WorkflowExecutionContext.TriggerActivityId); + return context.TransientProperties.GetOrAdd(GraphTransientProperty, () => new FlowGraph(flowchart.Connections, startActivity)); + } + + internal static async Task CancelInboundAncestorsAsync(this ActivityExecutionContext flowchartContext, IActivity activity) + { + if(flowchartContext.Activity is not Activities.Flowchart) + throw new InvalidOperationException("Activity context is not a flowchart."); + + var flowGraph = flowchartContext.GetFlowGraph(); + var ancestorActivities = flowGraph.GetAncestorActivities(activity); + var inboundActivityExecutionContexts = flowchartContext.WorkflowExecutionContext.ActivityExecutionContexts.Where(x => ancestorActivities.Contains(x.Activity) && x.ParentActivityExecutionContext == flowchartContext).ToList(); + + // Cancel each ancestor activity. + foreach (var activityExecutionContext in inboundActivityExecutionContexts) + { + await activityExecutionContext.CancelActivityAsync(); + } + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/ActivityExtensions.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/ActivityExtensions.cs new file mode 100644 index 000000000..3e26ce616 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/ActivityExtensions.cs @@ -0,0 +1,40 @@ +using Elsa.Extensions; +using Elsa.Workflows.Activities.Flowchart.Activities; +using Elsa.Workflows.Activities.Flowchart.Models; + +namespace Elsa.Workflows.Activities.Flowchart.Extensions; + +public static class ActivityExtensions +{ + public static MergeMode? GetMergeMode(this IActivity activity) + { + activity.CustomProperties.TryGetValue("mergeMode", out var mergeModeString); + return Enum.TryParse((string?)mergeModeString, true, out var mergeMode) ? mergeMode : null; + } + + public static void SetMergeMode(this IActivity activity, MergeMode? value) + { + if (value == null) + activity.CustomProperties.Remove("mergeMode"); + else + activity.CustomProperties["mergeMode"] = value; + } + + public static async Task GetMergeModeAsync(this IActivity activity, ActivityExecutionContext context) + { + if (activity.Type != "Elsa.FlowJoin") + { + return activity.GetMergeMode(); + } + + // Handle deprecated FlowJoin activity by evaluating its JoinMode property and mapping it to the appropriate MergeMode equivalent. + var joinActivityExecutionContext = await context.WorkflowExecutionContext.CreateActivityExecutionContextAsync(activity); + var joinMode = await joinActivityExecutionContext.EvaluateInputPropertyAsync(x => x.Mode); + + return joinMode switch + { + FlowJoinMode.WaitAny => MergeMode.Race, + _ => MergeMode.Converge + }; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/FlowchartExtensions.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/FlowchartExtensions.cs new file mode 100644 index 000000000..db026dd8e --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/FlowchartExtensions.cs @@ -0,0 +1,17 @@ +namespace Elsa.Workflows.Activities.Flowchart.Extensions; + +public static class FlowchartExtensions +{ + public static IActivity? GetRootActivity(this Activities.Flowchart flowchart) + { + // Get the first activity that has no inbound connections. + var query = + from activity in flowchart.Activities + let inboundConnections = flowchart.Connections.Any(x => x.Target.Activity == activity) + where !inboundConnections + select activity; + + var rootActivity = query.FirstOrDefault(); + return rootActivity; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/FlowGraph.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/FlowGraph.cs index a73e0ca32..5875ca50a 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/FlowGraph.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/FlowGraph.cs @@ -7,10 +7,11 @@ namespace Elsa.Workflows.Activities.Flowchart.Models; /// Represents a directed graph structure for managing workflow connections. /// Caches forward and backward connections to optimize graph traversal. /// -public class FlowGraph(ICollection Connections, IActivity? RootActivity) +public class FlowGraph(ICollection connections, IActivity? rootActivity) { private List? _cachedForwardConnections; private readonly Dictionary> _cachedInboundForwardConnections = new(); + private readonly Dictionary> _cachedInboundConnections = new(); private readonly Dictionary> _cachedOutboundConnections = new(); private readonly Dictionary _cachedIsBackwardConnection = new(); private readonly Dictionary _cachedIsDanglingActivity = new(); @@ -19,7 +20,7 @@ public class FlowGraph(ICollection Connections, IActivity? RootActiv /// /// Gets the list of forward connections, computing them if not already cached. /// - private List ForwardConnections => _cachedForwardConnections ??= RootActivity == null ? new() : GetForwardConnections(Connections, RootActivity); + private List ForwardConnections => _cachedForwardConnections ??= rootActivity == null ? new() : GetForwardConnections(connections, rootActivity); /// /// Retrieves all inbound forward connections for a given activity. @@ -29,12 +30,17 @@ public class FlowGraph(ICollection Connections, IActivity? RootActiv /// /// Retrieves all outbound connections for a given activity. /// - public List GetOutboundConnections(IActivity activity) => _cachedOutboundConnections.GetOrAdd(activity, () => Connections.OutboundConnections(activity).ToList()); + public List GetOutboundConnections(IActivity activity) => _cachedOutboundConnections.GetOrAdd(activity, () => connections.OutboundConnections(activity).ToList()); + + /// + /// Retrieves all inbound connections for a given activity. + /// + public List GetInboundConnections(IActivity activity) => _cachedInboundConnections.GetOrAdd(activity, () => connections.InboundConnections(activity).ToList()); /// /// Determines if a given activity is "dangling," meaning it does not exist as a target in any forward connection. /// - public bool IsDanglingActivity(IActivity activity) => _cachedIsDanglingActivity.GetOrAdd(activity, () => activity != RootActivity && !ForwardConnections.Any(c => c.Target.Activity == activity)); + public bool IsDanglingActivity(IActivity activity) => _cachedIsDanglingActivity.GetOrAdd(activity, () => activity != rootActivity && ForwardConnections.All(c => c.Target.Activity != activity)); /// /// Determines if a given connection is a backward connection (i.e., not part of the forward traversal) and whether it is valid. @@ -52,7 +58,7 @@ public class FlowGraph(ICollection Connections, IActivity? RootActiv bool isBackwardConnection = !GetForwardInboundConnections(connection.Target.Activity).Contains(connection); // Compute if the backward connection is valid - isValid = isBackwardConnection ? IsValidBackwardConnection(ForwardConnections, RootActivity, connection) : false; + isValid = isBackwardConnection && IsValidBackwardConnection(ForwardConnections, rootActivity, connection); // Cache the result _cachedIsBackwardConnection[connection] = (isBackwardConnection, isValid); @@ -108,7 +114,7 @@ public class FlowGraph(ICollection Connections, IActivity? RootActiv foreach (var conn in connections) { if (!adjList.ContainsKey(conn.Source.Activity)) - adjList[conn.Source.Activity] = new List(); + adjList[conn.Source.Activity] = new(); adjList[conn.Source.Activity].Add(conn.Target.Activity); } @@ -210,7 +216,8 @@ public class FlowGraph(ICollection Connections, IActivity? RootActiv { List> paths = new(); Queue> queue = new(); - queue.Enqueue(new List { start }); + queue.Enqueue(new() + { start }); while (queue.Count > 0) { @@ -219,7 +226,7 @@ public class FlowGraph(ICollection Connections, IActivity? RootActiv if (lastNode == root) { - paths.Add(new List(path)); + paths.Add([..path]); continue; } diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/MergeMode.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/MergeMode.cs new file mode 100644 index 000000000..28fdf0a03 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/MergeMode.cs @@ -0,0 +1,22 @@ +namespace Elsa.Workflows.Activities.Flowchart.Models; + +/// +/// Specifies the strategy for handling multiple inbound execution paths in a workflow. +/// +public enum MergeMode +{ + /// + /// Wait for all inbound paths before proceeding. + /// + Converge, + + /// + /// Proceed when any one inbound path completes; cancel all others. + /// + Race, + + /// + /// Proceed when any one inbound path completes; do not cancel others. + /// + Stream +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/Token.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/Token.cs new file mode 100644 index 000000000..465427ad3 --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/Token.cs @@ -0,0 +1,26 @@ +namespace Elsa.Workflows.Activities.Flowchart.Models; + +internal class Token(string fromActivityId, string? fromActivityName, string? outcome, string toActivityId, string? toActivityName, bool consumed, bool blocked) +{ + public static Token Create(IActivity from, IActivity to, string? outcome) => new(from.Id, from.Name, outcome, to.Id, to.Name, false, false); + + public string FromActivityId { get; } = fromActivityId; + public string? FromActivityName { get; } = fromActivityName; + public string? Outcome { get; } = outcome; + public string ToActivityId { get; } = toActivityId; + public string? ToActivityName { get; } = toActivityName; + public bool Consumed { get; private set; } = consumed; + public bool Blocked { get; private set; } = blocked; + + public Token Consume() + { + Consumed = true; + return this; + } + + public Token Block() + { + Blocked = true; + return this; + } +} \ 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 ada56a486..0254a95bb 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs @@ -551,7 +551,7 @@ public partial class WorkflowExecutionContext : IExecutionContext var now = SystemClock.UtcNow; var id = IdentityGenerator.GenerateId(); var activityExecutionContext = new ActivityExecutionContext(id, this, parentContext, activity, activityDescriptor, now, tag, SystemClock, CancellationToken); - var variablesToDeclare = options?.Variables ?? Array.Empty(); + var variablesToDeclare = options?.Variables ?? []; var variableContainer = new[] { activityExecutionContext.ActivityNode diff --git a/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs b/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs index 82eb8c22b..fc3574c33 100644 --- a/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs +++ b/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs @@ -127,6 +127,9 @@ public class FlowchartNextActivityTests [Fact(DisplayName = "Flowchart with an invalid backward connection")] public async Task InvalidBackwardConnectionTest() { + if(Flowchart.UseTokenFlow) + return; + var workflow = new TestWorkflow(workflowBuilder => { var start = new Start() { Id = "Start" };