Refactors and clarifies Flowchart merge modes (#6993)
* Refactors and clarifies Flowchart merge modes Improves the clarity and functionality of Flowchart merge modes by: - Renaming `None` to `Stream` for opportunistic execution. - Introducing `Merge` for waiting on activated branches only. - Enhancing `Converge` to be the strictest mode, requiring all inbound connections. - Providing more detailed descriptions for each mode, emphasizing their behavior and use cases in flow-based terminology. - Updates default merge mode to Stream This provides better control over synchronization and execution behavior in workflows. * Refactor `ActivityExtensions` to improve formatting, fix indentation, and align comments for improved readability and consistency * Update flowchart tests: replace `SetMergeMode(MergeMode.None)` with `SetMergeMode(null)` and remove unused `Elsa.Workflows` imports. * Restore None value for Flowchart MergeMode enum (#7107) * Initial plan * Add None value to MergeMode enum for backward compatibility Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Fix API client GetMergeMode to maintain non-nullable return type Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * [WIP] Address feedback on flowchart merge modes refactor (#7106) * Initial plan * Fix misleading documentation for Merge mode to match actual implementation Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Update test/integration/Elsa.Workflows.IntegrationTests/Scenarios/JoinBehaviors/ForkDecisionJoinTests.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update test/integration/Elsa.Workflows.IntegrationTests/Scenarios/JoinBehaviors/ForkDecisionJoinTests.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update test scenarios for implicit join behavior in `ForkDecisionJoinTests`. Updated file references for merge and stream join modes. --------- Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
a1d4e541fc
commit
730c01d9b0
|
|
@ -24,27 +24,38 @@ We refine the Flowchart's token-based execution logic (`OnChildCompletedTokenBas
|
|||
```csharp
|
||||
namespace Elsa.Workflows.Activities.Flowchart.Models;
|
||||
|
||||
/// <summary>
|
||||
/// Specifies the strategy for handling multiple inbound execution paths in a workflow.
|
||||
/// Uses flow-based terminology to describe merge behavior.
|
||||
/// </summary>
|
||||
public enum MergeMode
|
||||
{
|
||||
/// <summary>
|
||||
/// No special merging; use approximation that proceeds after all upstream sources complete, ignoring dead paths.
|
||||
/// Suitable for flexible, unstructured merges where optional branches shouldn't block.
|
||||
/// </summary>
|
||||
None,
|
||||
|
||||
/// <summary>
|
||||
/// Strict wait for tokens from all forward inbound connections. Blocks on dead/untaken paths.
|
||||
/// Use for required synchronization points.
|
||||
/// </summary>
|
||||
Converge,
|
||||
|
||||
/// <summary>
|
||||
/// Schedule on each arriving token, allowing multiple executions if supported.
|
||||
/// Flows freely when possible, ignoring dead/untaken paths.
|
||||
/// Opportunistic execution based on upstream completion.
|
||||
/// </summary>
|
||||
Stream,
|
||||
|
||||
/// <summary>
|
||||
/// Schedule on the first arriving token, block or cancel others.
|
||||
/// Merges only the activated/flowing inbound branches.
|
||||
/// Waits for all branches that received tokens, ignoring unactivated ones.
|
||||
/// </summary>
|
||||
Merge,
|
||||
|
||||
/// <summary>
|
||||
/// Converges all inbound paths, requiring every connection to execute.
|
||||
/// Strictest mode - will block on dead/untaken paths.
|
||||
/// </summary>
|
||||
Converge,
|
||||
|
||||
/// <summary>
|
||||
/// Cascades execution for each arriving token independently.
|
||||
/// Allows multiple concurrent executions (one per arriving token).
|
||||
/// </summary>
|
||||
Cascade,
|
||||
|
||||
/// <summary>
|
||||
/// Races inbound branches, executing on first arrival and blocking others.
|
||||
/// </summary>
|
||||
Race
|
||||
}
|
||||
|
|
@ -62,14 +73,14 @@ We refine the Flowchart's token-based execution logic (`OnChildCompletedTokenBas
|
|||
```csharp
|
||||
switch (mergeMode)
|
||||
{
|
||||
case MergeMode.Stream:
|
||||
case MergeMode.Cascade:
|
||||
case MergeMode.Race:
|
||||
// Existing logic: Schedule on arrival, block others for Race.
|
||||
// Schedule on arrival; for Race, block/cancel others.
|
||||
// ...
|
||||
break;
|
||||
|
||||
case MergeMode.Converge:
|
||||
// Strict check: Wait for all forward inbounds if >1; else schedule immediately.
|
||||
case MergeMode.Merge:
|
||||
// Wait for tokens from all forward inbound connections (activated branches only).
|
||||
var inboundConnections = flowGraph.GetForwardInboundConnections(targetActivity);
|
||||
if (inboundConnections.Count > 1)
|
||||
{
|
||||
|
|
@ -82,9 +93,23 @@ We refine the Flowchart's token-based execution logic (`OnChildCompletedTokenBas
|
|||
}
|
||||
break;
|
||||
|
||||
case MergeMode.None:
|
||||
case MergeMode.Converge:
|
||||
// Strictest mode: Wait for tokens from ALL inbound connections (forward + backward).
|
||||
var allInboundConnections = flowGraph.GetInboundConnections(targetActivity);
|
||||
if (allInboundConnections.Count > 1)
|
||||
{
|
||||
var hasAllTokens = allInboundConnections.All(inbound => /* token check */);
|
||||
if (hasAllTokens) await flowContext.ScheduleActivityAsync(...);
|
||||
}
|
||||
else
|
||||
{
|
||||
await flowContext.ScheduleActivityAsync(...);
|
||||
}
|
||||
break;
|
||||
|
||||
case MergeMode.Stream:
|
||||
default:
|
||||
// Approximation: Schedule if no unconsumed tokens to inbound sources.
|
||||
// Flows freely - approximation that proceeds when upstream completes.
|
||||
var inboundConnections = flowGraph.GetForwardInboundConnections(targetActivity);
|
||||
var hasUnconsumed = inboundConnections.Any(inbound => /* source token check */);
|
||||
if (!hasUnconsumed) await flowContext.ScheduleActivityAsync(...);
|
||||
|
|
@ -104,10 +129,10 @@ Flowchart execution starts with scheduling the root/start activity. As activitie
|
|||
This ensures acyclic forward flow with support for backward loops, using tokens to track control without global state beyond the list.
|
||||
|
||||
### Merge Modes Explained
|
||||
Each mode defines how tokens from multiple inbounds are synchronized:
|
||||
Each mode defines how tokens from multiple inbounds are synchronized using flow-based terminology:
|
||||
|
||||
- **None (Default/Flexible Merge)**:
|
||||
- **Behavior**: Schedules if there are no unconsumed tokens *to the sources* of inbounds (i.e., all upstream activities have completed, treating dead paths as "done").
|
||||
- **Stream (Flexible/Opportunistic Flow)**:
|
||||
- **Behavior**: Flows freely when possible, ignoring dead/untaken paths. Schedules if there are no unconsumed tokens *to the sources* of inbounds (i.e., all upstream activities have completed, treating dead paths as "done").
|
||||
- **When to Use**: Flexible merges in unstructured flows; optional/exclusive branches (e.g., switch defaults) shouldn't block.
|
||||
- **Scenarios**:
|
||||
- **Forks with Untaken Paths**: Proceeds after active branches (e.g., in complex switch with dangling default).
|
||||
|
|
@ -115,18 +140,26 @@ Each mode defines how tokens from multiple inbounds are synchronized:
|
|||
- **Dead Paths**: Ignores untaken outcomes; no blocking.
|
||||
- **Example**: In a switch with MatchAny, untaken default doesn't hang the merge.
|
||||
|
||||
- **Converge (Strict Synchronization)**:
|
||||
- **Behavior**: Requires unconsumed, non-blocked tokens from *all* forward inbounds. For <=1 forward, schedules immediately (loop/sequential friendly).
|
||||
- **When to Use**: Required "all must happen" joins; block if any branch untaken.
|
||||
- **Merge (Activated Branches Synchronization)**:
|
||||
- **Behavior**: Merges only activated/flowing inbound branches. Requires unconsumed, non-blocked tokens from *all* forward inbounds that received tokens. For <=1 forward, schedules immediately (loop/sequential friendly).
|
||||
- **When to Use**: Synchronization points where only activated paths matter; block if any activated branch hasn't completed.
|
||||
- **Scenarios**:
|
||||
- **Conditional Forks**: Blocks downstream if e.g., decision returns false and subsequent activities are connected to the true branch.
|
||||
- **Conditional Forks**: Blocks downstream if decision returns false and subsequent activities are connected to the true branch, but only waits for activated branches.
|
||||
- **Loops**: Works if forward inbounds <=1; reschedules on backward tokens.
|
||||
- **Dead Paths**: Blocks (desired for safety).
|
||||
- **Example**: Converge after parallel approvals—only proceed if all complete.
|
||||
- **Dead Paths**: Ignores untaken branches; waits only for activated ones.
|
||||
- **Example**: Merge after parallel approvals—only proceed if all activated approval branches complete.
|
||||
|
||||
- **Stream (Per-Token Execution)**:
|
||||
- **Behavior**: Schedules on each arriving token; may allow multiple concurrent executions of the target.
|
||||
- **When to Use**: Streaming merges where each branch triggers independently (e.g., event streams).
|
||||
- **Converge (Strictest - All Paths Required)**:
|
||||
- **Behavior**: Converges ALL inbound paths, requiring every connection to execute (forward AND backward). Most strict mode.
|
||||
- **When to Use**: When every single inbound path must execute before proceeding, regardless of activation status.
|
||||
- **Scenarios**:
|
||||
- **Strict Barriers**: Forces all possible paths to complete before proceeding.
|
||||
- **Dead Paths**: Blocks on dead/untaken paths (desired for maximum safety).
|
||||
- **Example**: Critical synchronization point requiring absolute completion of all defined paths.
|
||||
|
||||
- **Cascade (Per-Token Execution)**:
|
||||
- **Behavior**: Cascades execution for each arriving token independently; may allow multiple concurrent executions of the target.
|
||||
- **When to Use**: Streaming scenarios where each branch should trigger separate processing (e.g., event streams).
|
||||
- **Scenarios**:
|
||||
- **Forks**: Executes target per branch.
|
||||
- **Loops**: Executes per iteration.
|
||||
|
|
@ -134,8 +167,8 @@ Each mode defines how tokens from multiple inbounds are synchronized:
|
|||
- **Example**: Logging each branch outcome separately.
|
||||
|
||||
- **Race (First-Wins)**:
|
||||
- **Behavior**: Schedules on first token; blocks/cancels others (e.g., via blocked tokens and ancestor cancellation).
|
||||
- **When to Use**: Racing conditions (e.g., first response wins).
|
||||
- **Behavior**: Races inbound branches; schedules on first token, blocks/cancels others (e.g., via blocked tokens and ancestor cancellation).
|
||||
- **When to Use**: Racing conditions where first result wins (e.g., first response).
|
||||
- **Scenarios**:
|
||||
- **Forks**: Only first branch proceeds.
|
||||
- **Loops**: May race iterations if concurrent.
|
||||
|
|
@ -143,12 +176,12 @@ Each mode defines how tokens from multiple inbounds are synchronized:
|
|||
- **Example**: Waiting for fastest API response; cancel slower ones.
|
||||
|
||||
### Handling Common Scenarios
|
||||
|
||||
|
||||
- **Simple Sequential**: Any mode schedules on token arrival (single inbound).
|
||||
- **Fork-Join with Condition**: Converge blocks on false; None proceeds.
|
||||
- **Looping Construct**: All modes work; Converge uses count check to avoid strictness.
|
||||
- **Switch with Dangling Branches**: None ignores untaken; Converge blocks if required.
|
||||
- **BPMN Alignment**: None ≈ XOR/OR-join (flexible); Converge ≈ AND-join (strict); Race ≈ Event-based; Stream ≈ partial OR.
|
||||
- **Fork-Join with Condition**: Merge waits for activated branches; Converge blocks on all paths; Stream proceeds opportunistically.
|
||||
- **Looping Construct**: All modes work; Merge and Converge use count check to avoid strictness on single inbound.
|
||||
- **Switch with Dangling Branches**: Stream ignores untaken; Merge waits for activated; Converge blocks on all.
|
||||
- **BPMN Alignment**: Stream ≈ XOR/OR-join (flexible); Merge ≈ AND-join for active paths; Converge ≈ strict AND-join; Race ≈ Event-based; Cascade ≈ parallel multi-instance.
|
||||
|
||||
## Consequences
|
||||
|
||||
|
|
|
|||
|
|
@ -121,11 +121,16 @@ public static class ActivityExtensions
|
|||
|
||||
public static MergeMode GetMergeMode(this JsonObject activity)
|
||||
{
|
||||
return activity.GetProperty<MergeMode?>("customProperties", "mergeMode") ?? MergeMode.None;
|
||||
var value = activity.GetProperty<MergeMode?>("customProperties", "mergeMode");
|
||||
// Treat MergeMode.None as equivalent to null (no merge mode set), defaulting to Stream
|
||||
return value == null || value == MergeMode.None ? MergeMode.Stream : value.Value;
|
||||
}
|
||||
|
||||
public static void SetMergeMode(this JsonObject activity, MergeMode? value)
|
||||
{
|
||||
// Treat MergeMode.None as equivalent to null (no merge mode set)
|
||||
if (value == MergeMode.None)
|
||||
value = null;
|
||||
activity.SetProperty(JsonValue.Create(value), "customProperties", "mergeMode");
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -2,28 +2,51 @@ namespace Elsa.Api.Client.Shared.Enums;
|
|||
|
||||
/// <summary>
|
||||
/// Specifies the strategy for handling multiple inbound execution paths in a workflow.
|
||||
/// Uses flow-based terminology to describe merge behavior.
|
||||
/// </summary>
|
||||
public enum MergeMode
|
||||
{
|
||||
/// <summary>
|
||||
/// No special merging; use approximation that proceeds after all upstream sources complete, ignoring dead paths.
|
||||
/// Suitable for flexible, unstructured merges where optional branches shouldn't block.
|
||||
/// No merge mode set. Treated as if merge mode was not specified (null).
|
||||
/// Provides backward compatibility for existing workflows.
|
||||
/// Defaults to Stream behavior at runtime.
|
||||
/// </summary>
|
||||
None,
|
||||
|
||||
/// <summary>
|
||||
/// Strict wait for tokens from all forward inbound connections. Blocks on dead/untaken paths.
|
||||
/// Use for required synchronization points.
|
||||
/// </summary>
|
||||
Converge,
|
||||
|
||||
/// <summary>
|
||||
/// Schedule on each arriving token, allowing multiple executions if supported.
|
||||
/// Flows freely when possible, ignoring dead/untaken paths.
|
||||
/// Opportunistic execution based on upstream completion.
|
||||
/// Uses approximation that proceeds after all upstream sources complete.
|
||||
/// Suitable for flexible, unstructured merges where optional branches shouldn't block.
|
||||
/// </summary>
|
||||
Stream,
|
||||
|
||||
/// <summary>
|
||||
/// Merges only the activated/flowing inbound branches.
|
||||
/// Waits for all branches that received tokens, ignoring unactivated ones.
|
||||
/// Use for synchronization points where only taken paths matter (e.g., fork-joins with conditions).
|
||||
/// </summary>
|
||||
Merge,
|
||||
|
||||
/// <summary>
|
||||
/// Converges all inbound paths, requiring every connection to execute.
|
||||
/// Blocks until all branches complete, including unactivated ones.
|
||||
/// Strictest mode - will block on dead/untaken paths.
|
||||
/// Use when every single inbound path must execute before proceeding.
|
||||
/// </summary>
|
||||
Converge,
|
||||
|
||||
/// <summary>
|
||||
/// Cascades execution for each arriving token independently.
|
||||
/// Allows multiple concurrent executions (one per arriving token).
|
||||
/// Use for streaming scenarios where each branch should trigger separate processing.
|
||||
/// </summary>
|
||||
Cascade,
|
||||
|
||||
/// <summary>
|
||||
/// Races inbound branches, executing on first arrival and blocking others.
|
||||
/// Schedule on the first arriving token, block or cancel others.
|
||||
/// Use for competitive scenarios where only the first result matters.
|
||||
/// </summary>
|
||||
Race
|
||||
}
|
||||
|
|
@ -37,7 +37,7 @@ public partial class Flowchart
|
|||
|
||||
switch (mergeMode)
|
||||
{
|
||||
case MergeMode.Stream:
|
||||
case MergeMode.Cascade:
|
||||
case MergeMode.Race:
|
||||
if (mergeMode == MergeMode.Race)
|
||||
await flowContext.CancelInboundAncestorsAsync(targetActivity);
|
||||
|
|
@ -73,13 +73,15 @@ public partial class Flowchart
|
|||
|
||||
break;
|
||||
|
||||
case MergeMode.Converge:
|
||||
// Strict WaitAll for multiple forwards; schedule on arrival for <=1 (e.g., loops).
|
||||
var inboundConnectionsConverge = flowGraph.GetForwardInboundConnections(targetActivity);
|
||||
case MergeMode.Merge:
|
||||
// Wait for tokens from all forward inbound connections.
|
||||
// Unlike Converge, this ignores backward connections (loops).
|
||||
// Schedule on arrival for <=1 forward inbound (e.g., loops, sequential).
|
||||
var inboundConnectionsMerge = flowGraph.GetForwardInboundConnections(targetActivity);
|
||||
|
||||
if (inboundConnectionsConverge.Count > 1)
|
||||
if (inboundConnectionsMerge.Count > 1)
|
||||
{
|
||||
var hasAllTokens = inboundConnectionsConverge.All(inbound =>
|
||||
var hasAllTokens = inboundConnectionsMerge.All(inbound =>
|
||||
tokens.Any(t =>
|
||||
t is { Consumed: false, Blocked: false } &&
|
||||
t.FromActivityId == inbound.Source.Activity.Id &&
|
||||
|
|
@ -98,11 +100,37 @@ public partial class Flowchart
|
|||
|
||||
break;
|
||||
|
||||
case MergeMode.None:
|
||||
case MergeMode.Converge:
|
||||
// Strictest mode: Wait for tokens from ALL inbound connections (forward + backward).
|
||||
// Requires every possible inbound path to execute before proceeding.
|
||||
var allInboundConnectionsConverge = flowGraph.GetInboundConnections(targetActivity);
|
||||
|
||||
if (allInboundConnectionsConverge.Count > 1)
|
||||
{
|
||||
var hasAllTokens = allInboundConnectionsConverge.All(inbound =>
|
||||
tokens.Any(t =>
|
||||
t is { Consumed: false, Blocked: false } &&
|
||||
t.FromActivityId == inbound.Source.Activity.Id &&
|
||||
t.ToActivityId == targetActivity.Id &&
|
||||
t.Outcome == inbound.Source.Port
|
||||
)
|
||||
);
|
||||
|
||||
if (hasAllTokens)
|
||||
await flowContext.ScheduleActivityAsync(targetActivity, OnChildCompletedTokenBasedLogicAsync);
|
||||
}
|
||||
else
|
||||
{
|
||||
await flowContext.ScheduleActivityAsync(targetActivity, OnChildCompletedTokenBasedLogicAsync);
|
||||
}
|
||||
|
||||
break;
|
||||
|
||||
case MergeMode.Stream:
|
||||
default:
|
||||
// Approximation that proceeds on dead paths.
|
||||
var inboundConnectionsNone = flowGraph.GetForwardInboundConnections(targetActivity);
|
||||
var hasUnconsumed = inboundConnectionsNone.Any(inbound =>
|
||||
// Flows freely - approximation that proceeds when upstream completes, ignoring dead paths.
|
||||
var inboundConnectionsStream = flowGraph.GetForwardInboundConnections(targetActivity);
|
||||
var hasUnconsumed = inboundConnectionsStream.Any(inbound =>
|
||||
tokens.Any(t => !t.Consumed && !t.Blocked && t.ToActivityId == inbound.Source.Activity.Id)
|
||||
);
|
||||
|
||||
|
|
|
|||
|
|
@ -14,17 +14,21 @@ public static class ActivityExtensions
|
|||
return null;
|
||||
|
||||
// Handle both string and enum values for backwards compatibility
|
||||
return value switch
|
||||
var result = value switch
|
||||
{
|
||||
MergeMode mode => mode,
|
||||
string str when Enum.TryParse<MergeMode>(str, true, out var mode) => mode,
|
||||
_ => null
|
||||
_ => (MergeMode?)null
|
||||
};
|
||||
|
||||
// Treat MergeMode.None as equivalent to null (no merge mode set)
|
||||
return result == MergeMode.None ? null : result;
|
||||
}
|
||||
|
||||
public void SetMergeMode(MergeMode? value)
|
||||
{
|
||||
if (value == null)
|
||||
// Treat MergeMode.None as equivalent to null (no merge mode set)
|
||||
if (value == null || value == MergeMode.None)
|
||||
activity.CustomProperties.Remove("mergeMode");
|
||||
else
|
||||
activity.CustomProperties["mergeMode"] = value.ToString()!;
|
||||
|
|
@ -36,7 +40,7 @@ public static class ActivityExtensions
|
|||
{
|
||||
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<FlowJoin, FlowJoinMode>(x => x.Mode);
|
||||
|
|
@ -44,7 +48,7 @@ public static class ActivityExtensions
|
|||
return joinMode switch
|
||||
{
|
||||
FlowJoinMode.WaitAny => MergeMode.Race,
|
||||
_ => MergeMode.Converge
|
||||
_ => MergeMode.Merge // WaitAll maps to Merge (wait for all activated branches)
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,28 +2,52 @@ namespace Elsa.Workflows.Activities.Flowchart.Models;
|
|||
|
||||
/// <summary>
|
||||
/// Specifies the strategy for handling multiple inbound execution paths in a workflow.
|
||||
/// Uses flow-based terminology to describe merge behavior.
|
||||
/// </summary>
|
||||
public enum MergeMode
|
||||
{
|
||||
/// <summary>
|
||||
/// No special merging; use approximation that proceeds after all upstream sources complete, ignoring dead paths.
|
||||
/// Suitable for flexible, unstructured merges where optional branches shouldn't block.
|
||||
/// No merge mode set. Treated as if merge mode was not specified (null).
|
||||
/// Provides backward compatibility for existing workflows.
|
||||
/// Defaults to Stream behavior at runtime.
|
||||
/// </summary>
|
||||
None,
|
||||
|
||||
/// <summary>
|
||||
/// Strict wait for tokens from all forward inbound connections. Blocks on dead/untaken paths.
|
||||
/// Use for required synchronization points.
|
||||
/// </summary>
|
||||
Converge,
|
||||
|
||||
/// <summary>
|
||||
/// Schedule on each arriving token, allowing multiple executions if supported.
|
||||
/// Flows freely when possible, ignoring dead/untaken paths.
|
||||
/// Opportunistic execution based on upstream completion.
|
||||
/// Uses approximation that proceeds after all upstream sources complete.
|
||||
/// Suitable for flexible, unstructured merges where optional branches shouldn't block.
|
||||
/// </summary>
|
||||
Stream,
|
||||
|
||||
/// <summary>
|
||||
/// Waits for all forward (acyclic) inbound connections before proceeding.
|
||||
/// Unlike Converge, this ignores backward connections (loops).
|
||||
/// Will block on dead/untaken paths if they are forward connections.
|
||||
/// Use for synchronization points in structured fork-join patterns.
|
||||
/// </summary>
|
||||
Merge,
|
||||
|
||||
/// <summary>
|
||||
/// Converges all inbound paths, requiring every connection to execute.
|
||||
/// Blocks until all branches complete, including unactivated ones.
|
||||
/// Strictest mode - will block on dead/untaken paths.
|
||||
/// Use when every single inbound path must execute before proceeding.
|
||||
/// </summary>
|
||||
Converge,
|
||||
|
||||
/// <summary>
|
||||
/// Cascades execution for each arriving token independently.
|
||||
/// Allows multiple concurrent executions (one per arriving token).
|
||||
/// Use for streaming scenarios where each branch should trigger separate processing.
|
||||
/// </summary>
|
||||
Cascade,
|
||||
|
||||
/// <summary>
|
||||
/// Races inbound branches, executing on first arrival and blocking others.
|
||||
/// Schedule on the first arriving token, block or cancel others.
|
||||
/// Use for competitive scenarios where only the first result matters.
|
||||
/// </summary>
|
||||
Race
|
||||
}
|
||||
|
|
@ -1,5 +1,4 @@
|
|||
using Elsa.Testing.Shared;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.Activities.Flowchart.Activities;
|
||||
using Elsa.Workflows.Activities.Flowchart.Models;
|
||||
|
|
|
|||
|
|
@ -1,5 +1,4 @@
|
|||
using Elsa.Testing.Shared;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.Activities.Flowchart.Activities;
|
||||
using Elsa.Workflows.Activities.Flowchart.Extensions;
|
||||
|
|
@ -239,7 +238,7 @@ public class FlowchartTokenBasedTests : IDisposable
|
|||
var branch1 = new WriteLine("Branch1");
|
||||
var branch2 = new WriteLine("Branch2");
|
||||
var noneMode = new WriteLine("NoneMode");
|
||||
noneMode.SetMergeMode(MergeMode.None);
|
||||
noneMode.SetMergeMode(null);
|
||||
|
||||
var flowchart = new Flowchart
|
||||
{
|
||||
|
|
|
|||
|
|
@ -13,25 +13,31 @@ public class ForkDecisionJoinTests
|
|||
_services = new TestApplicationBuilder(testOutputHelper).WithCapturingTextWriter(_capturingTextWriter).Build();
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "The implicit join configured with None merge mode should execute.")]
|
||||
public async Task ImplicitJoinNoneShouldExecute()
|
||||
[Fact(DisplayName = "The implicit join configured with Stream merge mode should execute.")]
|
||||
public async Task ImplicitJoinStreamShouldExecute()
|
||||
{
|
||||
await RunAndAssert("fork-decision-join-none.json", ["A", "C"]);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "The implicit join configured with Converge merge mode should not execute.")]
|
||||
public async Task ImplicitJoinConvergeShouldNotExecute()
|
||||
|
||||
[Fact(DisplayName = "The implicit join configured with Merge mode should not execute (waits for activated branches only).")]
|
||||
public async Task ImplicitJoinMergeShouldNotExecute()
|
||||
{
|
||||
await RunAndAssert("fork-decision-join-converge.json", ["A"]);
|
||||
}
|
||||
|
||||
|
||||
[Fact(DisplayName = "The implicit join configured with Converge mode should block (strictest - requires ALL inbound connections).")]
|
||||
public async Task ImplicitJoinConvergeShouldBlock()
|
||||
{
|
||||
await RunAndAssert("fork-decision-join-converge-strict.json", ["A"]);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "The explicit join configured with WaitAll join mode should block.")]
|
||||
public async Task ExplicitJoinWaitAllShouldBlock()
|
||||
{
|
||||
await RunAndAssert("fork-decision-join-waitall.json", ["A", "C", "B", "D"]);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "An implicit join from the True and False branches should execute because the join mode is None and by default, all active branches are joined.")]
|
||||
[Fact(DisplayName = "An implicit join from the True and False branches should execute because the join mode is Stream and by default, all active branches are joined.")]
|
||||
public async Task ImplicitJoinFromBranchesShouldExecute()
|
||||
{
|
||||
// Populate registries.
|
||||
|
|
@ -42,7 +48,7 @@ public class ForkDecisionJoinTests
|
|||
|
||||
// Execute.
|
||||
var workflowState = await _services.RunWorkflowUntilEndAsync(workflowDefinition.DefinitionId);
|
||||
|
||||
|
||||
// Assert.
|
||||
Assert.Equal(WorkflowStatus.Finished, workflowState.Status);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,184 @@
|
|||
{
|
||||
"$schema": "https://elsaworkflows.io/schemas/workflow-definition/v3.0.0/schema.json",
|
||||
"id": "strict-converge-test",
|
||||
"definitionId": "strict-converge-001",
|
||||
"name": "Fork-Decision-Converge-Strict",
|
||||
"description": "Tests the strictest Converge mode that requires ALL inbound connections to execute",
|
||||
"createdAt": "2025-10-20T00:00:00.000000+00:00",
|
||||
"version": 1,
|
||||
"toolVersion": "3.6.0.0",
|
||||
"variables": [],
|
||||
"inputs": [],
|
||||
"outputs": [],
|
||||
"outcomes": [],
|
||||
"customProperties": {},
|
||||
"isReadonly": false,
|
||||
"isSystem": false,
|
||||
"isLatest": true,
|
||||
"isPublished": true,
|
||||
"options": {
|
||||
"autoUpdateConsumingWorkflows": false
|
||||
},
|
||||
"root": {
|
||||
"id": "flowchart-root",
|
||||
"nodeId": "Workflow1:flowchart-root",
|
||||
"name": "Flowchart1",
|
||||
"type": "Elsa.Flowchart",
|
||||
"version": 1,
|
||||
"customProperties": {
|
||||
"notFoundConnections": [],
|
||||
"canStartWorkflow": false,
|
||||
"runAsynchronously": false
|
||||
},
|
||||
"metadata": {},
|
||||
"activities": [
|
||||
{
|
||||
"condition": {
|
||||
"typeName": "Boolean",
|
||||
"expression": {
|
||||
"type": "JavaScript",
|
||||
"value": "return false;"
|
||||
}
|
||||
},
|
||||
"id": "decision-1",
|
||||
"nodeId": "Workflow1:flowchart-root:decision-1",
|
||||
"name": "FlowDecision1",
|
||||
"type": "Elsa.FlowDecision",
|
||||
"version": 1,
|
||||
"customProperties": {
|
||||
"canStartWorkflow": false,
|
||||
"runAsynchronously": false
|
||||
},
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"id": "start-1",
|
||||
"nodeId": "Workflow1:flowchart-root:start-1",
|
||||
"name": "Start1",
|
||||
"type": "Elsa.Start",
|
||||
"version": 1,
|
||||
"customProperties": {
|
||||
"canStartWorkflow": false,
|
||||
"runAsynchronously": false
|
||||
},
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"text": {
|
||||
"typeName": "String",
|
||||
"expression": {
|
||||
"type": "Literal",
|
||||
"value": "A"
|
||||
}
|
||||
},
|
||||
"id": "writeline-a",
|
||||
"nodeId": "Workflow1:flowchart-root:writeline-a",
|
||||
"name": "WriteLine1",
|
||||
"type": "Elsa.WriteLine",
|
||||
"version": 1,
|
||||
"customProperties": {
|
||||
"canStartWorkflow": false,
|
||||
"runAsynchronously": false
|
||||
},
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"text": {
|
||||
"typeName": "String",
|
||||
"expression": {
|
||||
"type": "Literal",
|
||||
"value": "B"
|
||||
}
|
||||
},
|
||||
"id": "writeline-b",
|
||||
"nodeId": "Workflow1:flowchart-root:writeline-b",
|
||||
"name": "WriteLine2",
|
||||
"type": "Elsa.WriteLine",
|
||||
"version": 1,
|
||||
"customProperties": {
|
||||
"canStartWorkflow": false,
|
||||
"runAsynchronously": false
|
||||
},
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"text": {
|
||||
"typeName": "String",
|
||||
"expression": {
|
||||
"type": "Literal",
|
||||
"value": "C"
|
||||
}
|
||||
},
|
||||
"id": "writeline-c",
|
||||
"nodeId": "Workflow1:flowchart-root:writeline-c",
|
||||
"name": "WriteLine3",
|
||||
"type": "Elsa.WriteLine",
|
||||
"version": 1,
|
||||
"customProperties": {
|
||||
"canStartWorkflow": false,
|
||||
"runAsynchronously": false,
|
||||
"mergeMode": "Converge"
|
||||
},
|
||||
"metadata": {}
|
||||
}
|
||||
],
|
||||
"variables": [],
|
||||
"connections": [
|
||||
{
|
||||
"source": {
|
||||
"activity": "writeline-a",
|
||||
"port": "Done"
|
||||
},
|
||||
"target": {
|
||||
"activity": "writeline-c",
|
||||
"port": "In"
|
||||
},
|
||||
"vertices": []
|
||||
},
|
||||
{
|
||||
"source": {
|
||||
"activity": "writeline-b",
|
||||
"port": "Done"
|
||||
},
|
||||
"target": {
|
||||
"activity": "writeline-c",
|
||||
"port": "In"
|
||||
},
|
||||
"vertices": []
|
||||
},
|
||||
{
|
||||
"source": {
|
||||
"activity": "start-1",
|
||||
"port": "Done"
|
||||
},
|
||||
"target": {
|
||||
"activity": "writeline-a",
|
||||
"port": "In"
|
||||
},
|
||||
"vertices": []
|
||||
},
|
||||
{
|
||||
"source": {
|
||||
"activity": "start-1",
|
||||
"port": "Done"
|
||||
},
|
||||
"target": {
|
||||
"activity": "decision-1",
|
||||
"port": "In"
|
||||
},
|
||||
"vertices": []
|
||||
},
|
||||
{
|
||||
"source": {
|
||||
"activity": "decision-1",
|
||||
"port": "True"
|
||||
},
|
||||
"target": {
|
||||
"activity": "writeline-b",
|
||||
"port": "In"
|
||||
},
|
||||
"vertices": []
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
|
@ -174,7 +174,7 @@
|
|||
"customProperties": {
|
||||
"canStartWorkflow": false,
|
||||
"runAsynchronously": false,
|
||||
"mergeMode": "Converge"
|
||||
"mergeMode": "Merge"
|
||||
},
|
||||
"metadata": {
|
||||
"designer": {
|
||||
|
|
|
|||
|
|
@ -174,7 +174,7 @@
|
|||
"customProperties": {
|
||||
"canStartWorkflow": false,
|
||||
"runAsynchronously": false,
|
||||
"mergeMode": "None"
|
||||
"mergeMode": "Stream"
|
||||
},
|
||||
"metadata": {
|
||||
"designer": {
|
||||
|
|
|
|||
Loading…
Reference in a new issue