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

Introduce token-centric Flowchart execution model with configurable MergeMode (#6632)

Browse source 
* Switch Flowchart to partial class and implement token-based logic

Refactored Flowchart to use a partial class structure for better modularity. Introduced a token-based execution option alongside the existing counter-based logic, enabling flexible activity scheduling strategies within the flowchart.

* Add token-based execution logic to Flowchart activities

Introduced token-based workflow execution for Flowcharts, enabling more granular control over activity scheduling. Refactored Start activity resolution and optimized connection handling within FlowGraph. Added extensions for improved activity query and execution management.

* Refactor token-based flowchart logic and add loopback detection

Replaced the legacy token-based flowchart logic with a cleaner, modular implementation. Introduced `LoopbackDetector` for identifying back-edges in workflows, enabling explicit handling of loopback connections for improved execution flow. Minor adjustments were also made for code consistency and clarity.

* Introduce token-centric execution model for Flowchart activity

Replaced the execution-count heuristic in Flowchart with a robust token-centric model, improving handling of loops, joins, and forks. Added join semantics via `JoinKind` and `IJoinHintProvider` to enable flexible configuration. Updated documentation and dependencies to reflect the new model.

* Refactor activity property handling and enhance JSON serialization.

Standardize key names in activity property management for consistency and improve error handling in `GetJoinKind` methods. Add support for enum serialization with `JsonStringEnumConverter` in JSON utilities.

* Refactor flowchart token handling for clarity and efficiency

Revised the flowchart token management logic to improve readability and maintainability. Simplified token creation, consumption, and scheduling while removing redundant code. Updated token structure and streamlined the flow execution process to handle dynamic and static connections effectively.

* Refactor Flowchart activity scheduling logic

Consolidated child activity scheduling into a single method, reducing duplication across token-based and counter-based workflows. Removed unused fields, redundant imports, and legacy logic to improve maintainability and clarity of the codebase.

* Update .gitignore to track Azurite data folder

Added `docker/azurite-data` to the ignore list to exclude local Azurite data files from being included in the repository. Ensures proper handling of temporary files related to Azure storage emulation.

* Refactor flowchart logic and extract activity context utilities

Reorganized flowchart logic by introducing reusable extension methods for activity context operations. Moved "HasPendingWork" and "HasFaultedChildren" methods into extensions to improve readability and maintainability. Updated related flowchart logic to utilize these methods for cleaner and more modular code.

* Refactor join logic and clean up Flowchart activities

Removed obsolete `JoinKind` enum and `IJoinHintProvider` interface, replacing them with the new `JoinMode` concept. Simplified flowchart logic by consolidating flow graph handling and refining activity execution extensions. Updated `FlowJoin` to recommend property-based configuration over explicit usage.

* Avoid invalid test execution when token flow is enabled.

The added check prevents the test from running when `Flowchart.UseTokenFlow` is enabled, as this scenario is incompatible. This ensures test reliability and avoids unnecessary execution.

* Refactor token-based logic in Flowchart activity.

Simplified token handling and introduced clearer logic for WaitAny and WaitAll joins. Improved efficiency by reducing redundant checks and streamlining activity scheduling. This ensures better maintainability and functionality in complex workflows.

* Refactor Token logic into its own model class

Moved the Token logic out of the Flowchart partial class to a standalone `Token` model for better modularity and clarity. Adjusted token consumption behavior to mutate the token in place rather than replacing it.

* Fix ancestor cancellation logic in Flowchart activities

Pass the correct activity to CancelInboundAncestorsAsync to ensure accurate ancestor determination and cancellation. Added validation to confirm the activity context is a flowchart to prevent invalid operations. These changes improve the reliability of flow control in workflows.

* Refactor Token model to add scheduling functionality

Replaced the `consumed` parameter with a `Scheduled` property and added a `Schedule` method for mark scheduling. This enhances the token model's flexibility, improving flowchart activity management.

* WIP on (no branch)

* Remove unused WaitAny guard logic from Flowchart activity.

The WaitAny guard logic and associated key have been commented out and effectively disabled. This cleanup simplifies the code, removing unnecessary operations that were no longer in use. The change ensures better maintainability and focuses on the core functionality.

* Add `MergeMode` enum and refactor flowchart merging logic

Introduced a `MergeMode` enum to define strategies for handling multiple inbound execution paths. Updated flowchart logic to replace the previous `FlowJoinMode` with `MergeMode`, and refactored related extensions to accommodate the new enum. This enhances clarity and flexibility in workflow merging behaviors.

* Mark `FlowJoin` obsolete and enhance merge mode handling

Added the `[Obsolete]` attribute to the `FlowJoin` activity and deprecated properties to encourage using `MergeMode`. Replaced synchronous `GetMergeMode` with an asynchronous `GetMergeModeAsync` to handle legacy `FlowJoin` mappings dynamically. Other minor changes include simplifying variable initialization in `WorkflowExecutionContext`.

* Mark FlowJoin activity as obsolete

The FlowJoin activity is now marked obsolete since all activities support the MergeMode property, eliminating the need for explicit join steps. This change includes an annotation update to clearly communicate its deprecated status.

* Refactor `OnActivityCanceledAsync` method location

Moved the `OnActivityCanceledAsync` method from the `Flowchart.Counters` file to the main `Flowchart` file for improved organization and readability. This ensures the method aligns better with its primary logic and related functionality.

* Refactor flowchart logic to improve cancellation handling

Refactored `OnActivityCanceledAsync` to handle token-based flow and legacy flow separately for better clarity and maintainability. Moved `CompleteIfNoPendingWorkAsync` to a single shared location and updated pending work checks to include unconsumed tokens and faulted activities. Simplified redundant logic and ensured consistency across methods.

* Remove consumed tokens after activity completion

This update ensures that consumed tokens pointing to the completed activity are purged from the token list. This helps maintain a clean state and prevents unnecessary token accumulation during workflow processing.

* Add support for blocking tokens in flowchart activities.

Introduced a `Blocked` property to the `Token` model to prevent scheduling activities prematurely in certain merge modes. Updated token processing logic to handle blocked tokens effectively and ensure proper scheduling or consumption. Simplified token management by removing redundant methods and cleaning up consumed tokens appropriately.

* Refactor Flowchart activity cancellation logic.

Streamlined cancellation logic by separating token and counter flow handling into dedicated methods. Removed redundant properties and logic, improving readability and maintainability. Simplified `OnChildCompletedAsync` and related methods for consistency.

* Refactor token-centric flowchart execution model.

Updated the flowchart ADR to clarify the token-centric approach, including `MergeMode` rules and scheduling logic. Enhanced the model to handle loops, forks, and resumable activities more robustly while improving state management and supporting cancellation. Updated sequence diagram for clarity.

* Remove JoinMode-related methods from ActivityExtensions

The GetJoinMode and SetJoinMode methods were removed as they are no longer needed. This cleanup reduces unused code and improves maintainability of the ActivityExtensions class.

* Remove unused JoinMode enum

The JoinMode enum was determined to be unnecessary and has been deleted to clean up the codebase. This helps reduce redundancy and improve maintainability.

* Remove unused LoopbackDetector class from Flowchart module

The LoopbackDetector class was removed because it is no longer used in the codebase. Its functionality appears to be obsolete or unnecessary for the current requirements of the Flowchart module.

* Update `UseTokenFlow` to be configurable

Made `UseTokenFlow` a static field to allow external configuration. This enables users to switch between the token flow and the old counter-based model as needed. Updated documentation accordingly.
This commit is contained in:
Sipke Schoorstra 2025-05-09 20:07:49 +02:00 committed by GitHub
parent a037ab873f
commit 3a1ddb650f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 812 additions and 418 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -78,5 +78,5 @@ unlist.sh
/artifacts
/docker/data/
docker/azurite-data
docker/docker-compose-datadog.yml

78
Directory.Packages.props
View file

@ -31,10 +31,10 @@
<PackageVersion Include="DistributedLock.Postgres" Version="1.3.0"/>
<PackageVersion Include="DistributedLock.Redis" Version="1.0.3"/>
<PackageVersion Include="Elastic.Clients.Elasticsearch" Version="8.17.1"/>
<PackageVersion Include="Elsa.Studio" Version="3.5.0-preview.925"/>
<PackageVersion Include="Elsa.Studio.Agents" Version="3.5.0-preview.925"/>
<PackageVersion Include="Elsa.Studio.Core.BlazorWasm" Version="3.5.0-preview.925"/>
<PackageVersion Include="Elsa.Studio.Login.BlazorWasm" Version="3.5.0-preview.925"/>
<PackageVersion Include="Elsa.Studio" Version="3.6.0-preview.979"/>
<PackageVersion Include="Elsa.Studio.Agents" Version="3.6.0-preview.979"/>
<PackageVersion Include="Elsa.Studio.Core.BlazorWasm" Version="3.6.0-preview.979"/>
<PackageVersion Include="Elsa.Studio.Login.BlazorWasm" Version="3.6.0-preview.979"/>
<PackageVersion Include="FastEndpoints" Version="5.35.0"/>
<PackageVersion Include="FastEndpoints.Security" Version="5.35.0"/>
<PackageVersion Include="FastEndpoints.Swagger" Version="5.35.0"/>
@ -123,37 +123,37 @@
<PackageVersion Include="xunit.runner.visualstudio" Version="3.0.2"/>
<PackageVersion Include="Yarp.ReverseProxy" Version="2.3.0"/>
<PackageVersion Include="AspNetCore.Authentication.ApiKey" Version="8.0.1"/>
<PackageVersion Include="Microsoft.AspNetCore.Authorization" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Components" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly.DevServer" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.DataProtection.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Data.Sqlite" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Data.Sqlite.Core" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Relational" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Sqlite" Version="9.0.3"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.SqlServer" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Caching.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Configuration" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Http" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Http.Polly" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Logging" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Options" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="9.0.3"/>
<PackageVersion Include="MySql.Data" Version="9.2.0"/>
<PackageVersion Include="Microsoft.AspNetCore.Authorization" Version="9.0.4"/>
<PackageVersion Include="Microsoft.AspNetCore.Components" Version="9.0.4"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly" Version="9.0.4"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly.DevServer" Version="9.0.4"/>
<PackageVersion Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="9.0.4"/>
<PackageVersion Include="Microsoft.AspNetCore.DataProtection.Abstractions" Version="9.0.4"/>
<PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Data.Sqlite" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Data.Sqlite.Core" Version="9.0.4"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore" Version="9.0.4"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="9.0.4"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Relational" Version="9.0.4"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.Sqlite" Version="9.0.4"/>
<PackageVersion Include="Microsoft.EntityFrameworkCore.SqlServer" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Caching.Abstractions" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Configuration" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Http" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Http.Polly" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Logging" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Options" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="9.0.4"/>
<PackageVersion Include="MySql.Data" Version="9.3.0"/>
<PackageVersion Include="Npgsql" Version="9.0.3"/>
<PackageVersion Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="9.0.4"/>
<PackageVersion Include="Oracle.EntityFrameworkCore" Version="9.23.60"/>
@ -161,9 +161,9 @@
<PackageVersion Include="Pomelo.EntityFrameworkCore.MySql" Version="9.0.0-preview.3.efcore.9.0.0"/>
<PackageVersion Include="Refit" Version="8.0.0"/>
<PackageVersion Include="Refit.HttpClientFactory" Version="8.0.0"/>
<PackageVersion Include="System.Formats.Asn1" Version="9.0.3"/>
<PackageVersion Include="System.Text.Json" Version="9.0.3"/>
<PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="9.3.0"/>
<PackageVersion Include="Microsoft.Extensions.Resilience" Version="9.3.0"/>
<PackageVersion Include="System.Formats.Asn1" Version="9.0.4"/>
<PackageVersion Include="System.Text.Json" Version="9.0.4"/>
<PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="9.4.0"/>
<PackageVersion Include="Microsoft.Extensions.Resilience" Version="9.4.0"/>
</ItemGroup>
</Project>

1
Elsa.sln
View file

@ -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

83
doc/adr/0004-token-centric-flowchart-execution-model.md Normal file
View file

@ -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
```

13
src/clients/Elsa.Api.Client/Extensions/ActivityExtensions.cs
View file

@ -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.
/// </summary>
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<MergeMode?>("customProperties", "mergeMode") ?? MergeMode.Converge;
}
public static void SetMergeMode(this JsonObject activity, MergeMode? value)
{
activity.SetProperty(JsonValue.Create(value), "customProperties", "mergeMode");
}
/// <summary>
/// Gets the activities in the specified flowchart.
/// </summary>
public static IEnumerable<JsonObject> GetActivities(this JsonObject flowchart) => flowchart.GetProperty("activities")?.AsArray().AsEnumerable().Cast<JsonObject>() ?? Array.Empty<JsonObject>();
public static IEnumerable<JsonObject> GetActivities(this JsonObject flowchart) => flowchart.GetProperty("activities")?.AsArray().AsEnumerable().Cast<JsonObject>() ?? [];
/// <summary>
/// Sets the activities in the specified flowchart.

8
src/clients/Elsa.Api.Client/Extensions/JsonObjectExtensions.cs
View file

@ -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
/// <returns>A <see cref="JsonObject"/> representing the specified value.</returns>
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)!;
}

22
src/clients/Elsa.Api.Client/Shared/Enums/MergeMode.cs Normal file
View file

@ -0,0 +1,22 @@
namespace Elsa.Api.Client.Shared.Enums;
/// <summary>
/// Specifies the strategy for handling multiple inbound execution paths in a workflow.
/// </summary>
public enum MergeMode
{
/// <summary>
/// Wait for all inbound paths before proceeding.
/// </summary>
Converge,
/// <summary>
/// Proceed when any one inbound path completes; cancel all others.
/// </summary>
Race,
/// <summary>
/// Proceed when any one inbound path completes; do not cancel others.
/// </summary>
Stream
}

2
src/modules/Elsa.Workflows.Core/Activities/Container.cs
View file

@ -9,7 +9,7 @@ namespace Elsa.Workflows.Activities;
public abstract class Container : Activity, IVariableContainer
{
/// <inheritdoc />
protected Container(string? source = default, int? line = default) : base(source, line)
protected Container(string? source = null, int? line = null) : base(source, line)
{
}

21
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs
View file

@ -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;
/// <summary>
/// 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.
/// </summary>
[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
{
/// <inheritdoc />
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
/// <inheritdoc />
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
};
}
}

2
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowSwitch.cs
View file

@ -19,7 +19,7 @@ namespace Elsa.Workflows.Activities.Flowchart.Activities;
public class FlowSwitch : Activity
{
/// <inheritdoc />
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)
{
}

260
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Counters.cs Normal file
View file

@ -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<bool>(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());
}
/// <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 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;
}
/// <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 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<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, OnChildCompletedCounterBasedLogicAsync);
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;
}
}
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);
}
}

114
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.Tokens.cs Normal file
View file

@ -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<Token> GetTokenList(ActivityExecutionContext context)
{
if (context.Properties.TryGetValue(TokenStoreKey, out var obj) && obj is List<Token> list)
return list;
var newList = new List<Token>();
context.Properties[TokenStoreKey] = newList;
return newList;
}
}

385
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/Flowchart.cs
View file

@ -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;
/// </summary>
[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";
/// <summary>
/// Set this to <c>false</c> from your program file in case you wish to use the old counter based model.
/// </summary>
public static bool UseTokenFlow = true;
/// <inheritdoc />
public Flowchart([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
@ -41,7 +40,7 @@ public class Flowchart : Container
/// <inheritdoc />
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();
}
/// <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 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<bool>(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);
}
}
/// <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)
{
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;
}
/// <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 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<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;
}
}
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();
}
}

106
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/ActivityExecutionContextExtensions.cs Normal file
View file

@ -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();
}
/// <summary>
/// Checks if there is any pending work for the flowchart.
/// </summary>
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();
}
}
}

40
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/ActivityExtensions.cs Normal file
View file

@ -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<MergeMode>((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<MergeMode?> 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<FlowJoin, FlowJoinMode>(x => x.Mode);
return joinMode switch
{
FlowJoinMode.WaitAny => MergeMode.Race,
_ => MergeMode.Converge
};
}
}

17
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Extensions/FlowchartExtensions.cs Normal file
View file

@ -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;
}
}

23
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/FlowGraph.cs
View file

@ -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.
/// </summary>
public class FlowGraph(ICollection<Connection> Connections, IActivity? RootActivity)
public class FlowGraph(ICollection<Connection> connections, IActivity? rootActivity)
{
private List<Connection>? _cachedForwardConnections;
private readonly Dictionary<IActivity, List<Connection>> _cachedInboundForwardConnections = new();
private readonly Dictionary<IActivity, List<Connection>> _cachedInboundConnections = new();
private readonly Dictionary<IActivity, List<Connection>> _cachedOutboundConnections = new();
private readonly Dictionary<Connection, (bool IsBackwardConnection, bool IsValid)> _cachedIsBackwardConnection = new();
private readonly Dictionary<IActivity, bool> _cachedIsDanglingActivity = new();
@ -19,7 +20,7 @@ public class FlowGraph(ICollection<Connection> Connections, IActivity? RootActiv
/// <summary>
/// Gets the list of forward connections, computing them if not already cached.
/// </summary>
private List<Connection> ForwardConnections => _cachedForwardConnections ??= RootActivity == null ? new() : GetForwardConnections(Connections, RootActivity);
private List<Connection> ForwardConnections => _cachedForwardConnections ??= rootActivity == null ? new() : GetForwardConnections(connections, rootActivity);
/// <summary>
/// Retrieves all inbound forward connections for a given activity.
@ -29,12 +30,17 @@ public class FlowGraph(ICollection<Connection> Connections, IActivity? RootActiv
/// <summary>
/// Retrieves all outbound connections for a given activity.
/// </summary>
public List<Connection> GetOutboundConnections(IActivity activity) => _cachedOutboundConnections.GetOrAdd(activity, () => Connections.OutboundConnections(activity).ToList());
public List<Connection> GetOutboundConnections(IActivity activity) => _cachedOutboundConnections.GetOrAdd(activity, () => connections.OutboundConnections(activity).ToList());
/// <summary>
/// Retrieves all inbound connections for a given activity.
/// </summary>
public List<Connection> GetInboundConnections(IActivity activity) => _cachedInboundConnections.GetOrAdd(activity, () => connections.InboundConnections(activity).ToList());
/// <summary>
/// Determines if a given activity is "dangling," meaning it does not exist as a target in any forward connection.
/// </summary>
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));
/// <summary>
/// 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<Connection> 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<Connection> Connections, IActivity? RootActiv
foreach (var conn in connections)
{
if (!adjList.ContainsKey(conn.Source.Activity))
adjList[conn.Source.Activity] = new List<IActivity>();
adjList[conn.Source.Activity] = new();
adjList[conn.Source.Activity].Add(conn.Target.Activity);
}
@ -210,7 +216,8 @@ public class FlowGraph(ICollection<Connection> Connections, IActivity? RootActiv
{
List<List<IActivity>> paths = new();
Queue<List<IActivity>> queue = new();
queue.Enqueue(new List<IActivity> { start });
queue.Enqueue(new()
{ start });
while (queue.Count > 0)
{
@ -219,7 +226,7 @@ public class FlowGraph(ICollection<Connection> Connections, IActivity? RootActiv
if (lastNode == root)
{
paths.Add(new List<IActivity>(path));
paths.Add([..path]);
continue;
}

22
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/MergeMode.cs Normal file
View file

@ -0,0 +1,22 @@
namespace Elsa.Workflows.Activities.Flowchart.Models;
/// <summary>
/// Specifies the strategy for handling multiple inbound execution paths in a workflow.
/// </summary>
public enum MergeMode
{
/// <summary>
/// Wait for all inbound paths before proceeding.
/// </summary>
Converge,
/// <summary>
/// Proceed when any one inbound path completes; cancel all others.
/// </summary>
Race,
/// <summary>
/// Proceed when any one inbound path completes; do not cancel others.
/// </summary>
Stream
}

26
src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/Token.cs Normal file
View file

@ -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;
}
}

2
src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs
View file

@ -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<Variable>();
var variablesToDeclare = options?.Variables ?? [];
var variableContainer = new[]
{
activityExecutionContext.ActivityNode

3
test/integration/Elsa.Workflows.IntegrationTests/Scenarios/FlowchartNextActivity/Tests.cs
View file

@ -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" };