diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Execute/PostEndpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Execute/PostEndpoint.cs index 6148de9e6..a7afdcfe4 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Execute/PostEndpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Execute/PostEndpoint.cs @@ -29,19 +29,24 @@ internal class PostEndpoint( { PostRequest? request = null; - if (HttpContext.Request.ContentLength > 0 && (HttpContext.Request.ContentType?.Contains("application/json") ?? true)) + if (HttpContext.Request.ContentType?.Contains("application/json") ?? false) { - try + using var reader = new StreamReader(HttpContext.Request.Body); + var body = await reader.ReadToEndAsync(); + + if (!string.IsNullOrWhiteSpace(body)) { - request = await JsonSerializer.DeserializeAsync(HttpContext.Request.Body, - new JsonSerializerOptions + try { - PropertyNameCaseInsensitive = true - }, cancellationToken: cancellationToken); - } - catch - { - AddError("Invalid request body."); + request = JsonSerializer.Deserialize(body, new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true + }); + } + catch + { + AddError("Invalid request body."); + } } } 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 e5b868ab4..412e23c38 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs @@ -52,6 +52,343 @@ public partial class Flowchart : Container 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("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); + var 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) + { + 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($"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); + + if(!connectionFollowed) + continue; // Skip if connection was not followed. + + var outboundActivity = outboundConnection.Target.Activity; + + // Determine the 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($"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; + } + } + + var scheduleWorkOptions = new ScheduleWorkOptions + { + CompletionCallback = OnChildCompletedAsync, + ExistingActivityExecutionContext = joinContext + }; + await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions); + return true; + } + + 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; diff --git a/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs b/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs index ff3639ac3..b606285cb 100644 --- a/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs +++ b/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs @@ -32,7 +32,10 @@ public class FlowchartNextActivityTests await _services.PopulateRegistriesAsync(); await _workflowRunner.RunAsync(); var lines = _capturingTextWriter.Lines.ToList(); - Assert.Equal(new[] { "Line 1" }, lines); + Assert.Equal(new[] + { + "Line 1" + }, lines); } [Fact(DisplayName = "Flowchart with backward connections and a dangling activity")] @@ -54,7 +57,6 @@ public class FlowchartNextActivityTests }; var a = new WriteLine("A"); var b = new WriteLine("B"); - var c = new WriteLine("C"); var incrementLoop = new SetVariable() { Variable = loopVariable, @@ -73,28 +75,26 @@ public class FlowchartNextActivityTests var f = new WriteLine("F"); var end = new End(); - workflowBuilder.Root = new Flowchart { Variables = - { - loopVariable - }, + { + loopVariable + }, Activities = - { - start, - dangling, - writeLineDecision, - a, - b, - c, - incrementLoop, - loopbackDecision, - d, - e, - f, - end - }, + { + start, + dangling, + writeLineDecision, + a, + b, + incrementLoop, + loopbackDecision, + d, + e, + f, + end + }, Connections = { new(start, writeLineDecision), @@ -105,7 +105,6 @@ public class FlowchartNextActivityTests new(new(writeLineDecision, "Default"), new Endpoint(incrementLoop)), new(a, incrementLoop), new(b, incrementLoop), - new(c, incrementLoop), new(incrementLoop, loopbackDecision), new(new(loopbackDecision, "EqualOne"), new Endpoint(d)), new(d, incrementLoop), @@ -121,7 +120,10 @@ public class FlowchartNextActivityTests var result = await _workflowRunner.RunAsync(workflow); var lines = _capturingTextWriter.Lines.ToList(); Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus); - Assert.Equal(new[] { "A", "B", "C", "D", "E", "A", "B", "E", "F" }, lines); + Assert.Equal(new[] + { + "A", "B", "D", "E", "A", "B", "E", "E", "F" + }, lines); } [Fact(DisplayName = "Flowchart with an invalid backward connection")] @@ -132,24 +134,42 @@ public class FlowchartNextActivityTests var workflow = new TestWorkflow(workflowBuilder => { - var start = new Start() { Id = "Start" }; - var a = new WriteLine("A") { Id = "WriteLineA" }; - var b = new WriteLine("B") { Id = "WriteLineB" }; - var c = new WriteLine("C") { Id = "WriteLineC" }; - var d = new WriteLine("D") { Id = "WriteLineD" }; - var e = new WriteLine("E") { Id = "WriteLineE" }; + var start = new Start() + { + Id = "Start" + }; + var a = new WriteLine("A") + { + Id = "WriteLineA" + }; + var b = new WriteLine("B") + { + Id = "WriteLineB" + }; + var c = new WriteLine("C") + { + Id = "WriteLineC" + }; + var d = new WriteLine("D") + { + Id = "WriteLineD" + }; + var e = new WriteLine("E") + { + Id = "WriteLineE" + }; workflowBuilder.Root = new Flowchart { Activities = - { - start, - a, - b, - c, - d, - e, - }, + { + start, + a, + b, + c, + d, + e, + }, Connections = { new(start, a), @@ -169,7 +189,10 @@ public class FlowchartNextActivityTests Assert.Equal(WorkflowSubStatus.Faulted, result.WorkflowState.SubStatus); Assert.Equal(1, result.WorkflowState.Incidents.Count()); Assert.Equal("Invalid backward connection: Every path from the source ('WriteLineE') must go through the target ('WriteLineC') when tracing back to the start.", result.WorkflowState.Incidents.First().Message); - Assert.Equal(new[] { "A", "B", "C", "D", "E" }, lines); + Assert.Equal(new[] + { + "A", "B", "C", "D", "E" + }, lines); } [Theory(DisplayName = "Flowchart with a Join activity executed multiple times")] @@ -207,21 +230,21 @@ public class FlowchartNextActivityTests workflowBuilder.Root = new Flowchart { Variables = - { - loopVariable - }, + { + loopVariable + }, Activities = - { - start, - a, - b, - c, - d, - join, - incrementLoop, - loopbackDecision, - end - }, + { + start, + a, + b, + c, + d, + join, + incrementLoop, + loopbackDecision, + end + }, Connections = { new(start, a), @@ -279,19 +302,19 @@ public class FlowchartNextActivityTests workflowBuilder.Root = new Flowchart { Variables = - { - loopVariable - }, + { + loopVariable + }, Activities = - { - start, - loopbackSwitch, - a, - incrementLoop, - join, - b, - end - }, + { + start, + loopbackSwitch, + a, + incrementLoop, + join, + b, + end + }, Connections = { new(start, loopbackSwitch), @@ -310,6 +333,9 @@ public class FlowchartNextActivityTests var result = await _workflowRunner.RunAsync(workflow); var lines = _capturingTextWriter.Lines.ToList(); Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus); - Assert.Equal(new[] { "A", "A", "A", "B" }, lines); + Assert.Equal(new[] + { + "A", "A", "A", "B" + }, lines); } } \ No newline at end of file