Merge remote-tracking branch 'origin/patch/3.5.2' into develop/3.6.0

This commit is contained in:
Matt 2025-10-03 22:17:01 +01:00
commit 60d01f8892
3 changed files with 442 additions and 74 deletions

View file

@ -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<PostRequest>(HttpContext.Request.Body,
new JsonSerializerOptions
try
{
PropertyNameCaseInsensitive = true
}, cancellationToken: cancellationToken);
}
catch
{
AddError("Invalid request body.");
request = JsonSerializer.Deserialize<PostRequest>(body, new JsonSerializerOptions
{
PropertyNameCaseInsensitive = true
});
}
catch
{
AddError("Invalid request body.");
}
}
}

View file

@ -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();
}
/// <summary>
/// Checks if there is any pending work for the flowchart.
/// </summary>
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<bool>(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);
}
}
/// <summary>
/// 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.
/// </summary>
/// <param name="flowchart">The flowchart containing the activities.</param>
/// <param name="flowGraph">The graph representation of the flowchart.</param>
/// <param name="flowScope">Tracks activity and connection visits.</param>
/// <param name="flowchartContext">The execution context of the flowchart.</param>
/// <param name="activity">The current activity being processed.</param>
/// <param name="outcomes">The outcomes that determine which connections were followed.</param>
/// <param name="completedActivityExecutedByBackwardConnection">Indicates if the completed activity was executed due to a backward connection.</param>
/// <returns>True if at least one activity was scheduled; otherwise, false.</returns>
private async ValueTask<bool> 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;
}
/// <summary>
/// Schedules an outbound activity that originates from a backward connection.
/// </summary>
private async ValueTask<bool> 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<string, object>() { { BackwardConnectionActivityInput, true } }
};
await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions);
return true;
}
/// <summary>
/// Schedules a non-join activity if all its forward inbound connections have been visited.
/// </summary>
private async ValueTask<bool> 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);
}
}
/// <summary>
/// Schedules a join activity based on inbound connection statuses.
/// </summary>
private async ValueTask<bool> 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;

View file

@ -32,7 +32,10 @@ public class FlowchartNextActivityTests
await _services.PopulateRegistriesAsync();
await _workflowRunner.RunAsync<FlowchartWorkflow>();
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);
}
}