Merge branch 'develop/3.6.0' into feat/unit-test-coverage-flowjoin

src/apps/Elsa.Server.Web/Elsa.Server.Web.csproj

@@ -7,6 +7,7 @@
 		<ProjectReference Include="..\..\modules\Elsa.Persistence.EFCore.Sqlite\Elsa.Persistence.EFCore.Sqlite.csproj" />
 		<ProjectReference Include="..\..\modules\Elsa.Scheduling\Elsa.Scheduling.csproj" />
 		<ProjectReference Include="..\..\modules\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj" />
+		<ProjectReference Include="..\..\modules\Elsa.Workflows.Runtime.Distributed\Elsa.Workflows.Runtime.Distributed.csproj" />
 		<ProjectReference Include="..\..\modules\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
 		<ProjectReference Include="..\..\modules\Elsa\Elsa.csproj" />
 		<ProjectReference Include="..\..\modules\Elsa.Expressions.CSharp\Elsa.Expressions.CSharp.csproj" />

src/apps/Elsa.Server.Web/Program.cs

@@ -17,6 +17,7 @@ using Elsa.Workflows.CommitStates.Strategies;
 using Elsa.Workflows.IncidentStrategies;
 using Elsa.Workflows.LogPersistence;
 using Elsa.Workflows.Options;
+using Elsa.Workflows.Runtime.Distributed.Extensions;
 using Elsa.Workflows.Runtime.Options;
 using Elsa.Workflows.Runtime.Tasks;
 using JetBrains.Annotations;
@@ -70,6 +71,7 @@ services
             {
                 runtime.UseEntityFrameworkCore(ef => ef.UseSqlite());
                 runtime.UseCache();
+                runtime.UseDistributedRuntime();
             })
             .UseWorkflowsApi()
             .UseScheduling()

src/apps/Elsa.Server.Web/appsettings.json

@@ -2,7 +2,8 @@
   "Logging": {
     "LogLevel": {
       "Default": "Warning",
-      "Microsoft.Hosting.Lifetime": "Information"
+      "Microsoft.Hosting.Lifetime": "Information",
+      "Elsa": "Debug"
     }
   },
   "HostBuilder": {

src/common/Elsa.Testing.Shared.Component/Activities/TriggerSignal.cs

@@ -4,9 +4,9 @@ using Elsa.Workflows.Runtime;
 
 namespace Elsa.Testing.Shared.Activities;
 
-public class TriggerSignal(object signal) : CodeActivity
+public class TriggerSignal(string signal) : CodeActivity
 {
-    public object Signal { get; set; } = signal;
+    public string Signal { get; set; } = signal;
 
     protected override void Execute(ActivityExecutionContext context)
     {

src/common/Elsa.Testing.Shared.Component/Models/TestWorkflowExecutionResult.cs

@@ -1,6 +1,7 @@
+using Elsa.Workflows;
 using Elsa.Workflows.Runtime.Entities;
 
-namespace Elsa.Workflows.ComponentTests.Models;
+namespace Elsa.Testing.Shared.Models;
 
 /// <summary>
 /// Represents the result of a test workflow execution, including the workflow execution context and activity execution records.

src/common/Elsa.Testing.Shared.Component/Services/AsyncWorkflowRunner.cs

@@ -1,13 +1,12 @@
-using Elsa.Testing.Shared;
-using Elsa.Testing.Shared.Services;
-using Elsa.Workflows.ComponentTests.Models;
+using System.Collections.Concurrent;
+using Elsa.Testing.Shared.Models;
+using Elsa.Workflows;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Entities;
 using Elsa.Workflows.Runtime.Messages;
-using System.Collections.Concurrent;
 
-namespace Elsa.Workflows.ComponentTests.Services;
+namespace Elsa.Testing.Shared.Services;
 
 /// <summary>
 /// Provides functionality to execute workflows asynchronously and await their completion for testing purposes.

src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceVariableManager.cs

@@ -1,5 +1,11 @@
+using Elsa.Workflows.Management.Entities;
+using Elsa.Workflows.State;
+
 namespace Elsa.Workflows.Management;
 
+/// <summary>
+/// Defines the operations for managing variables associated with a workflow instance.
+/// </summary>
 public interface IWorkflowInstanceVariableManager
 {
     /// <summary>
@@ -8,8 +14,26 @@ public interface IWorkflowInstanceVariableManager
     /// <param name="workflowInstanceId">The ID of the workflow instance.</param>
     /// <param name="excludeTags"></param>
     /// <param name="cancellationToken">The cancellation token to cancel the operation.</param>
-    /// <returns>A task that represents the asynchronous operation. The task result contains a collection of <see cref="ResolvedVariable"/> instances.</returns>
-    Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(string workflowInstanceId, IEnumerable<string>? excludeTags = default, CancellationToken cancellationToken = default);
+    /// <returns>A collection of <see cref="ResolvedVariable"/> instances.</returns>
+    Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(string workflowInstanceId, IEnumerable<string>? excludeTags = null, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Retrieves all variables for the specified workflow instance.
+    /// </summary>
+    /// <param name="workflowInstance">The workflow instance from which to retrieve variables.</param>
+    /// <param name="excludeTags">A collection of tags to exclude from the variable results, if any.</param>
+    /// <param name="cancellationToken">The cancellation token to cancel the operation.</param>
+    /// <returns>A collection of <see cref="ResolvedVariable"/> instances.</returns>
+    Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(WorkflowInstance workflowInstance, IEnumerable<string>? excludeTags = null, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Retrieves all variables for the specified workflow state.
+    /// </summary>
+    /// <param name="workflowState">The workflow state to retrieve variables from.</param>
+    /// <param name="excludeTags">Optional tags to exclude from the result.</param>
+    /// <param name="cancellationToken">The cancellation token to cancel the operation.</param>
+    /// <returns>A collection of <see cref="ResolvedVariable"/> instances.</returns>
+    Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(WorkflowState workflowState, IEnumerable<string>? excludeTags = null, CancellationToken cancellationToken = default);
 
     /// <summary>
     /// Retrieves all variables from the specified <see cref="WorkflowExecutionContext"/>.
@@ -17,8 +41,8 @@ public interface IWorkflowInstanceVariableManager
     /// <param name="workflowExecutionContext">The context of the workflow execution.</param>
     /// <param name="excludeTags"></param>
     /// <param name="cancellationToken">The cancellation token to cancel the operation.</param>
-    /// <returns>A task that represents the asynchronous operation. The task result contains a collection of <see cref="ResolvedVariable"/> instances.</returns>
-    Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(WorkflowExecutionContext workflowExecutionContext, IEnumerable<string>? excludeTags = default, CancellationToken cancellationToken = default);
+    /// <returns>A collection of <see cref="ResolvedVariable"/> instances.</returns>
+    Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(WorkflowExecutionContext workflowExecutionContext, IEnumerable<string>? excludeTags = null, CancellationToken cancellationToken = default);
     
     /// <summary>
     /// Sets the specified variables in the specified workflow instance.

src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceVariableManager.cs

@@ -1,24 +1,39 @@
+using Elsa.Workflows.Management.Entities;
+using Elsa.Workflows.State;
+
 namespace Elsa.Workflows.Management.Services;
 
 public class WorkflowInstanceVariableManager(
-    IWorkflowInstanceManager workflowInstanceManager, 
-    IWorkflowDefinitionService workflowDefinitionService, 
-    IServiceProvider serviceProvider, 
+    IWorkflowInstanceManager workflowInstanceManager,
+    IWorkflowDefinitionService workflowDefinitionService,
+    IServiceProvider serviceProvider,
     IWorkflowInstanceVariableReader variableReader,
     IWorkflowInstanceVariableWriter variableWriter) : IWorkflowInstanceVariableManager
 {
-    public async Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(string workflowInstanceId, IEnumerable<string>? excludeTags = default, CancellationToken cancellationToken = default)
+    public async Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(string workflowInstanceId, IEnumerable<string>? excludeTags = null, CancellationToken cancellationToken = default)
     {
         var workflowExecutionContext = await GetWorkflowExecutionContextAsync(workflowInstanceId, cancellationToken);
         if (workflowExecutionContext == null) return [];
         return await variableReader.GetVariables(workflowExecutionContext, excludeTags, cancellationToken);
     }
 
-    public Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(WorkflowExecutionContext workflowExecutionContext, IEnumerable<string>? excludeTags = default, CancellationToken cancellationToken = default)
+    public Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(WorkflowExecutionContext workflowExecutionContext, IEnumerable<string>? excludeTags = null, CancellationToken cancellationToken = default)
     {
         return variableReader.GetVariables(workflowExecutionContext, excludeTags, cancellationToken);
     }
 
+    public async Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(WorkflowInstance workflowInstance, IEnumerable<string>? excludeTags = null, CancellationToken cancellationToken = default)
+    {
+        return await GetVariablesAsync(workflowInstance.WorkflowState, excludeTags, cancellationToken);
+    }
+
+    public async Task<IEnumerable<ResolvedVariable>> GetVariablesAsync(WorkflowState workflowState, IEnumerable<string>? excludeTags = null, CancellationToken cancellationToken = default)
+    {
+        var workflowExecutionContext = await GetWorkflowExecutionContextAsync(workflowState, cancellationToken);
+        if (workflowExecutionContext == null) return [];
+        return await variableReader.GetVariables(workflowExecutionContext, excludeTags, cancellationToken);
+    }
+
     public async Task<IEnumerable<ResolvedVariable>> SetVariablesAsync(string workflowInstanceId, IEnumerable<VariableUpdateValue> variables, CancellationToken cancellationToken = default)
     {
         var workflowExecutionContext = await GetWorkflowExecutionContextAsync(workflowInstanceId, cancellationToken);
@@ -32,20 +47,26 @@ public class WorkflowInstanceVariableManager(
     {
         return variableWriter.SetVariables(workflowExecutionContext, variables, cancellationToken);
     }
-    
+
     private async Task<WorkflowExecutionContext?> GetWorkflowExecutionContextAsync(string workflowInstanceId, CancellationToken cancellationToken)
     {
         var workflowInstance = await workflowInstanceManager.FindByIdAsync(workflowInstanceId, cancellationToken);
 
         if (workflowInstance == null)
             return null;
-        
+
         var workflowState = workflowInstance.WorkflowState;
+
+        return await GetWorkflowExecutionContextAsync(workflowState, cancellationToken);
+    }
+
+    private async Task<WorkflowExecutionContext?> GetWorkflowExecutionContextAsync(WorkflowState workflowState, CancellationToken cancellationToken)
+    {
         var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowState.DefinitionVersionId, cancellationToken);
-        
+
         if (workflowGraph == null)
             return null;
-        
+
         return await WorkflowExecutionContext.CreateAsync(
             serviceProvider,
             workflowGraph,

src/modules/Elsa.Workflows.Runtime/Providers/ClrWorkflowsProvider.cs

@@ -51,6 +51,6 @@ public class ClrWorkflowsProvider(
         };
 
         var materializerContext = new ClrWorkflowMaterializerContext(workflowBuilder.GetType());
-        return new MaterializedWorkflow(workflow, Name, ClrWorkflowMaterializer.MaterializerName, materializerContext);
+        return new(workflow, Name, ClrWorkflowMaterializer.MaterializerName, materializerContext);
     }
 }

test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs

@@ -12,7 +12,6 @@ using Elsa.Testing.Shared.Handlers;
 using Elsa.Testing.Shared.Services;
 using Elsa.Workflows.ComponentTests.Decorators;
 using Elsa.Workflows.ComponentTests.Materializers;
-using Elsa.Workflows.ComponentTests.Services;
 using Elsa.Workflows.ComponentTests.WorkflowProviders;
 using Elsa.Workflows.Management;
 using Elsa.Workflows.Runtime.Distributed.Extensions;

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs

@@ -0,0 +1,177 @@
+using Elsa.Common.Models;
+using Elsa.Testing.Shared;
+using Elsa.Testing.Shared.Models;
+using Elsa.Testing.Shared.Services;
+using Elsa.Workflows.Activities;
+using Elsa.Workflows.ComponentTests.Abstractions;
+using Elsa.Workflows.ComponentTests.Fixtures;
+using Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+using Elsa.Workflows.Management;
+using Elsa.Workflows.Models;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows;
+
+public class BulkDispatchWorkflowsTests : AppComponentTest
+{
+    private readonly AsyncWorkflowRunner _workflowRunner;
+
+    public BulkDispatchWorkflowsTests(App app) : base(app)
+    {
+        _workflowRunner = Scope.ServiceProvider.GetRequiredService<AsyncWorkflowRunner>();
+    }
+
+    [Fact(DisplayName = "BulkDispatchWorkflows should wait for all child workflows to complete")]
+    public async Task BulkDispatchAndWait_ShouldWaitForAllChildWorkflowsToComplete()
+    {
+        var result = await RunWorkflowAsync(BulkDispatchAndWaitWorkflow.DefinitionId);
+
+        var writeLineExecutionRecords = result.ActivityExecutionRecords.Where(x => x.ActivityType == "Elsa.WriteLine").ToList();
+        Assert.Equal(4, writeLineExecutionRecords.Count);
+    }
+
+    [Fact(DisplayName = "BulkDispatchWorkflows should dispatch and not wait when WaitForCompletion is false")]
+    public async Task BulkDispatchFireAndForget_ShouldNotWaitForChildWorkflows()
+    {
+        var expectedChildCount = 3;
+
+        // Run the main workflow and wait for child workflows to complete
+        var (result, completedChildWorkflows) = await RunWorkflowAndWaitForChildWorkflowsAsync(
+            BulkDispatchFireAndForgetWorkflow.DefinitionId,
+            SlowBulkChildWorkflow.DefinitionId,
+            expectedChildCount);
+
+        AssertWorkflowFinished(result);
+        var mainWorkflowCompletedAt = result.WorkflowExecutionContext.UpdatedAt;
+
+        // Assert that all child workflows completed after the main workflow
+        Assert.Equal(expectedChildCount, completedChildWorkflows.Count);
+        foreach (var childContext in completedChildWorkflows)
+        {
+            Assert.True(childContext.UpdatedAt > mainWorkflowCompletedAt,
+                $"Child workflow should complete after main workflow. Main: {mainWorkflowCompletedAt}, Child: {childContext.UpdatedAt}");
+        }
+    }
+
+    [Fact(DisplayName = "BulkDispatchWorkflows should use CorrelationIdFunction")]
+    public async Task BulkDispatchWithCorrelationId_ShouldUseCorrelationIdFunction()
+    {
+        var expectedChildCount = 3;
+
+        // Run the main workflow and wait for child workflows to complete
+        var (result, completedChildWorkflows) = await RunWorkflowAndWaitForChildWorkflowsAsync(
+            BulkDispatchWithCorrelationIdWorkflow.DefinitionId,
+            BulkChildWorkflow.DefinitionId,
+            expectedChildCount);
+
+        AssertWorkflowFinished(result);
+
+        // Assert that all child workflows have the expected correlation IDs based on the CorrelationIdFunction
+        Assert.Equal(expectedChildCount, completedChildWorkflows.Count);
+
+        var expectedCorrelationIds = new[] { "correlation-1", "correlation-2", "correlation-3" };
+        var actualCorrelationIds = completedChildWorkflows.Select(c => c.CorrelationId).OrderBy(c => c).ToList();
+
+        Assert.Equal(expectedCorrelationIds, actualCorrelationIds);
+    }
+
+    [Fact(DisplayName = "BulkDispatchWorkflows should execute ChildFaulted ports")]
+    public async Task BulkDispatchWithChildPorts_ShouldExecuteChildFaultedPortForFaultedWorkflows()
+    {
+        var result = await RunWorkflowAsync(BulkDispatchWithChildPortsWorkflow.DefinitionId);
+        AssertWorkflowFinished(result);
+
+        var faultedCount = await GetWorkflowVariableAsync<int>(result, "FaultedCount");
+        Assert.Equal(3, faultedCount);
+    }
+
+    [Fact(DisplayName = "BulkDispatchWorkflows should complete immediately when Items is empty")]
+    public async Task BulkDispatchWithEmptyItems_ShouldCompleteImmediately()
+    {
+        var result = await RunWorkflowAsync(BulkDispatchEmptyItemsWorkflow.DefinitionId);
+        AssertWorkflowFinished(result);
+    }
+
+    [Fact(DisplayName = "BulkDispatchWorkflows should throw when workflow definition not found")]
+    public async Task BulkDispatchWithInvalidWorkflowDefinitionId_ShouldThrow()
+    {
+        var result = await RunWorkflowAsync(BulkDispatchInvalidDefinitionWorkflow.DefinitionId);
+        Assert.Equal(WorkflowSubStatus.Faulted, result.WorkflowExecutionContext.SubStatus);
+    }
+
+    [Fact(DisplayName = "BulkDispatchWorkflows child workflows should receive current item")]
+    public async Task BulkDispatchWorkflows_ChildWorkflowsShouldReceiveCurrentItem()
+    {
+        var result = await RunWorkflowAsync(MixFruitsWorkflow.DefinitionId);
+        AssertWorkflowFinished(result);
+
+        var writeLineExecutionRecords = result.ActivityExecutionRecords.Where(x => x.ActivityType == "Elsa.WriteLine").ToList();
+        Assert.Equal(3, writeLineExecutionRecords.Count);
+
+        var writtenTexts = writeLineExecutionRecords
+            .Select(x => x.ActivityState?[nameof(WriteLine.Text)] as string)
+            .ToList();
+
+        Assert.Contains("Mixing Apple", writtenTexts);
+        Assert.Contains("Mixing Banana", writtenTexts);
+        Assert.Contains("Mixing Cherry", writtenTexts);
+    }
+
+    private Task<TestWorkflowExecutionResult> RunWorkflowAsync(string workflowDefinitionId)
+    {
+        return _workflowRunner.RunAndAwaitWorkflowCompletionAsync(WorkflowDefinitionHandle.ByDefinitionId(workflowDefinitionId, VersionOptions.Published));
+    }
+
+    private static void AssertWorkflowFinished(TestWorkflowExecutionResult result)
+    {
+        Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowExecutionContext.SubStatus);
+    }
+
+    private async Task<T> GetWorkflowVariableAsync<T>(TestWorkflowExecutionResult result, string variableName)
+    {
+        var variableManager = Scope.ServiceProvider.GetRequiredService<IWorkflowInstanceVariableManager>();
+        var variables = await variableManager.GetVariablesAsync(result.WorkflowExecutionContext);
+        return (T?)variables.FirstOrDefault(v => v.Variable.Name == variableName)?.Value;
+    }
+
+    private async Task<(TestWorkflowExecutionResult Result, List<WorkflowExecutionContext> CompletedChildWorkflows)> RunWorkflowAndWaitForChildWorkflowsAsync(
+        string parentWorkflowDefinitionId,
+        string childWorkflowDefinitionId,
+        int expectedChildCount)
+    {
+        var workflowEvents = Scope.ServiceProvider.GetRequiredService<WorkflowEvents>();
+        var completedChildWorkflows = new List<WorkflowExecutionContext>();
+        var childWorkflowCompletionTcs = new TaskCompletionSource();
+
+        // Subscribe to child workflow completion events
+        void OnWorkflowStateCommitted(object? sender, WorkflowStateCommittedEventArgs e)
+        {
+            if (e.WorkflowExecutionContext.Workflow.Identity.DefinitionId != childWorkflowDefinitionId ||
+                e.WorkflowExecutionContext.Status != WorkflowStatus.Finished)
+            {
+                return;
+            }
+
+            completedChildWorkflows.Add(e.WorkflowExecutionContext);
+            if (completedChildWorkflows.Count == expectedChildCount)
+                childWorkflowCompletionTcs.TrySetResult();
+        }
+
+        workflowEvents.WorkflowStateCommitted += OnWorkflowStateCommitted;
+
+        try
+        {
+            // Run the main workflow
+            var result = await RunWorkflowAsync(parentWorkflowDefinitionId);
+
+            // Wait for all child workflows to complete
+            await childWorkflowCompletionTcs.Task;
+
+            return (result, completedChildWorkflows);
+        }
+        finally
+        {
+            workflowEvents.WorkflowStateCommitted -= OnWorkflowStateCommitted;
+        }
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/BulkChildWorkflow.cs

@@ -1,12 +1,11 @@
 using Elsa.Extensions;
 using Elsa.Workflows.Activities;
-using Elsa.Workflows.ComponentTests.Activities;
 using JetBrains.Annotations;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.BulkDispatchWorkflows.Workflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
 
 [UsedImplicitly]
-public class FruitWorkflow : WorkflowBase
+public class BulkChildWorkflow : WorkflowBase
 {
     public static readonly string DefinitionId = Guid.NewGuid().ToString();
 
@@ -14,13 +13,13 @@ public class FruitWorkflow : WorkflowBase
     {
         builder.WithDefinitionId(DefinitionId);
         var item = builder.WithInput<string>("Item");
+
         builder.Root = new Sequence
         {
             Activities =
             {
-                new WriteLine(x => $"Mixing {x.GetInput<string>(item)}"),
-                new TriggerSignal(x => x.GetInput<string>(item))
+                new WriteLine(context => $"Processing item: {context.GetInput<string>(item)}")
             }
         };
     }
-}
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/BulkDispatchAndWaitWorkflow.cs

@@ -0,0 +1,26 @@
+using Elsa.Workflows.Activities;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+public class BulkDispatchAndWaitWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString();
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+        builder.Root = new Sequence
+        {
+            Activities =
+            {
+                new Runtime.Activities.BulkDispatchWorkflows
+                {
+                    WorkflowDefinitionId = new(BulkChildWorkflow.DefinitionId),
+                    Items = new(new object[] { 1, 2, 3 }),
+                    WaitForCompletion = new(true)
+                },
+                new WriteLine("Done")
+            }
+        };
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/BulkDispatchEmptyItemsWorkflow.cs

@@ -0,0 +1,21 @@
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+[UsedImplicitly]
+public class BulkDispatchEmptyItemsWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = nameof(BulkDispatchEmptyItemsWorkflow);
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+
+        builder.Root = new Runtime.Activities.BulkDispatchWorkflows
+        {
+            WorkflowDefinitionId = new(BulkChildWorkflow.DefinitionId),
+            Items = new(Array.Empty<object>()),
+            WaitForCompletion = new(true)
+        };
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/BulkDispatchFireAndForgetWorkflow.cs

@@ -0,0 +1,17 @@
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+public class BulkDispatchFireAndForgetWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString();
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+        builder.Root = new Runtime.Activities.BulkDispatchWorkflows
+        {
+            WorkflowDefinitionId = new(SlowBulkChildWorkflow.DefinitionId),
+            Items = new(new object[] { "A", "B", "C" }),
+            WaitForCompletion = new(false)
+        };
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/BulkDispatchInvalidDefinitionWorkflow.cs

@@ -0,0 +1,31 @@
+using Elsa.Workflows.Activities;
+using Elsa.Workflows.IncidentStrategies;
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+[UsedImplicitly]
+public class BulkDispatchInvalidDefinitionWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = nameof(BulkDispatchInvalidDefinitionWorkflow);
+    public static readonly string InvalidChildWorkflowId = "NonExistentWorkflow";
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+        builder.WorkflowOptions.IncidentStrategyType = typeof(FaultStrategy);
+
+        builder.Root = new Sequence
+        {
+            Activities =
+            {
+                new Runtime.Activities.BulkDispatchWorkflows
+                {
+                    WorkflowDefinitionId = new(InvalidChildWorkflowId),
+                    Items = new(() => new object[] { 1 }),
+                    WaitForCompletion = new(true)
+                }
+            }
+        };
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/BulkDispatchWithChildPortsWorkflow.cs

@@ -0,0 +1,46 @@
+using Elsa.Extensions;
+using Elsa.Workflows.Activities;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+public class BulkDispatchWithChildPortsWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString();
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+
+        var completedCountVariable = builder.WithVariable("CompletedCount", 0).WithWorkflowStorage();
+        var faultedCountVariable = builder.WithVariable("FaultedCount", 0).WithWorkflowStorage();
+
+        builder.Root = new Runtime.Activities.BulkDispatchWorkflows
+        {
+            WorkflowDefinitionId = new(FaultingChildWorkflow.DefinitionId),
+            Items = new(new object[] { 1, 2, 3 }),
+            WaitForCompletion = new(true),
+            ChildCompleted = new Sequence
+            {
+                Activities =
+                {
+                    new SetVariable
+                    {
+                        Variable = completedCountVariable,
+                        Value = new(context => completedCountVariable.Get(context) + 1)
+                    }
+                }
+            },
+            ChildFaulted = new Sequence
+            {
+                Activities =
+                {
+                    new SetVariable
+                    {
+                        Variable = faultedCountVariable,
+                        Value = new(context => faultedCountVariable.Get(context) + 1)
+                    }
+                }
+            }
+        };
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/BulkDispatchWithCorrelationIdWorkflow.cs

@@ -0,0 +1,23 @@
+using Elsa.Expressions.JavaScript.Models;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+public class BulkDispatchWithCorrelationIdWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString();
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+
+        var items = new object[] { 1, 2, 3 };
+
+        builder.Root = new Runtime.Activities.BulkDispatchWorkflows
+        {
+            WorkflowDefinitionId = new(BulkChildWorkflow.DefinitionId),
+            Items = new(items),
+            CorrelationIdFunction = new(JavaScriptExpression.Create("`correlation-${getItem()}`")),
+            WaitForCompletion = new(true)
+        };
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/EmployeeGreetingWorkflow.cs

@@ -2,7 +2,7 @@ using Elsa.Extensions;
 using Elsa.Workflows.Activities;
 using JetBrains.Annotations;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.BulkDispatchWorkflows.Workflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
 
 [UsedImplicitly]
 public class EmployeeGreetingWorkflow : WorkflowBase

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/FaultingChildWorkflow.cs

@@ -0,0 +1,28 @@
+using Elsa.Workflows.Activities;
+using Elsa.Workflows.IncidentStrategies;
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+[UsedImplicitly]
+public class FaultingChildWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString();
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+        builder.WorkflowOptions.IncidentStrategyType = typeof(FaultStrategy);
+
+        builder.Root = new Sequence
+        {
+            Activities =
+            {
+                new Fault
+                {
+                    Message = new("Child workflow failed")
+                }
+            }
+        };
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/FruitWorkflow.cs

@@ -0,0 +1,18 @@
+using Elsa.Extensions;
+using Elsa.Workflows.Activities;
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+[UsedImplicitly]
+public class FruitWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString();
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+        var item = builder.WithInput<string>("Item");
+        builder.Root = new WriteLine(x => $"Mixing {x.GetInput<string>(item)}");
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/GreetEmployeesWorkflow.cs

@@ -1,7 +1,7 @@
 using Elsa.Testing.Shared.Activities;
 using Elsa.Workflows.Activities;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.BulkDispatchWorkflows.Workflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
 
 public class GreetEmployeesWorkflow : WorkflowBase
 {

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/MixFruitsWorkflow.cs

@@ -1,6 +1,6 @@
 using Elsa.Workflows.Activities;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.BulkDispatchWorkflows.Workflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
 
 public class MixFruitsWorkflow : WorkflowBase
 {

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/BulkDispatchWorkflows/Workflows/SlowBulkChildWorkflow.cs

@@ -0,0 +1,27 @@
+using Elsa.Extensions;
+using Elsa.Scheduling.Activities;
+using Elsa.Workflows.Activities;
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.BulkDispatchWorkflows.Workflows;
+
+[UsedImplicitly]
+public class SlowBulkChildWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString();
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+        var item = builder.WithInput<string>("Item");
+
+        builder.Root = new Sequence
+        {
+            Activities =
+            {
+                Delay.FromMilliseconds(10),
+                new WriteLine(context => $"Processing item: {context.GetInput<string>(item)}")
+            }
+        };
+    }
+}

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/DispatchWorkflows/DispatchWorkflowsTests.cs

@@ -3,7 +3,7 @@ using Elsa.Testing.Shared;
 using Elsa.Testing.Shared.Services;
 using Elsa.Workflows.ComponentTests.Abstractions;
 using Elsa.Workflows.ComponentTests.Fixtures;
-using Elsa.Workflows.ComponentTests.Scenarios.DispatchWorkflows.Workflows;
+using Elsa.Workflows.ComponentTests.Scenarios.Activities.DispatchWorkflows.Workflows;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Messages;

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/DispatchWorkflows/Workflows/ChildWorkflow.cs

@@ -2,7 +2,7 @@ using Elsa.Scheduling.Activities;
 using Elsa.Workflows.Activities;
 using JetBrains.Annotations;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.DispatchWorkflows.Workflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.DispatchWorkflows.Workflows;
 
 [UsedImplicitly]
 public class ChildWorkflow : WorkflowBase

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/DispatchWorkflows/Workflows/DispatchAndWaitWorkflow.cs

@@ -2,7 +2,7 @@ using Elsa.Testing.Shared.Activities;
 using Elsa.Workflows.Activities;
 using Elsa.Workflows.Runtime.Activities;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.DispatchWorkflows.Workflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.DispatchWorkflows.Workflows;
 
 public class DispatchAndWaitWorkflow : WorkflowBase
 {

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/ExecuteWorkflows/ExecuteWorkflowsTests.cs

@@ -1,13 +1,13 @@
 using Elsa.Common.Models;
 using Elsa.Workflows.ComponentTests.Abstractions;
 using Elsa.Workflows.ComponentTests.Fixtures;
-using Elsa.Workflows.ComponentTests.Scenarios.ExecuteWorkflows.Workflows;
+using Elsa.Workflows.ComponentTests.Scenarios.Activities.ExecuteWorkflows.Workflows;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Messages;
 using Microsoft.Extensions.DependencyInjection;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.ExecuteWorkflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.ExecuteWorkflows;
 
 public class ExecuteWorkflowsTests : AppComponentTest
 {
@@ -22,7 +22,7 @@ public class ExecuteWorkflowsTests : AppComponentTest
     public async Task ExecuteWorkflow_ShouldExecuteWorkflow()
     {
         var workflowClient = await _workflowRuntime.CreateClientAsync();
-        await workflowClient.CreateInstanceAsync(new CreateWorkflowInstanceRequest
+        await workflowClient.CreateInstanceAsync(new()
         {
             WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(MainWorkflow.DefinitionId, VersionOptions.Published)
         });

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/ExecuteWorkflows/Workflows/MainWorkflow.cs

@@ -3,7 +3,7 @@ using Elsa.Workflows.Activities;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Activities;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.ExecuteWorkflows.Workflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.ExecuteWorkflows.Workflows;
 
 public class MainWorkflow : WorkflowBase
 {

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/ExecuteWorkflows/Workflows/SubroutineWorkflow.cs

@@ -3,7 +3,7 @@ using Elsa.Workflows.Activities;
 using Elsa.Workflows.Management.Activities.SetOutput;
 using JetBrains.Annotations;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.ExecuteWorkflows.Workflows;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.ExecuteWorkflows.Workflows;
 
 [UsedImplicitly]
 public class SubroutineWorkflow : WorkflowBase

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/FlowJoin/Tests.cs

@@ -2,7 +2,7 @@
 using Elsa.Workflows.ComponentTests.Fixtures;
 using Microsoft.Extensions.DependencyInjection;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.FlowJoins;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.FlowJoin;
 
 public class Tests(App app) : AppComponentTest(app)
 {

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/FlowJoin/Workflows.cs

@@ -1,16 +1,16 @@
 using Elsa.Workflows.Activities.Flowchart.Activities;
 
-namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.FlowJoins;
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.FlowJoin;
 
 public class SingleJoinWorkflow : WorkflowBase
 {
     protected override void Build(IWorkflowBuilder builder)
     {
-        builder.Root = new Elsa.Workflows.Activities.Flowchart.Activities.Flowchart
+        builder.Root = new Flowchart
         {
             Activities =
             {
-                new FlowJoin()
+                new Workflows.Activities.Flowchart.Activities.FlowJoin()
             }
         };
     }

test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/ForEach/ForEachWorkflowTests.cs

@@ -1,8 +1,8 @@
 using Elsa.Common.Models;
+using Elsa.Testing.Shared.Services;
 using Elsa.Workflows.ComponentTests.Abstractions;
 using Elsa.Workflows.ComponentTests.Fixtures;
 using Elsa.Workflows.ComponentTests.Scenarios.Activities.ForEach.Workflows;
-using Elsa.Workflows.ComponentTests.Services;
 using Elsa.Workflows.Models;
 using Microsoft.Extensions.DependencyInjection;
 

test/component/Elsa.Workflows.ComponentTests/Scenarios/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs

@@ -1,56 +0,0 @@
-using Elsa.Common.Models;
-using Elsa.Testing.Shared.Services;
-using Elsa.Workflows.ComponentTests.Abstractions;
-using Elsa.Workflows.ComponentTests.Fixtures;
-using Elsa.Workflows.ComponentTests.Scenarios.BulkDispatchWorkflows.Workflows;
-using Elsa.Workflows.Models;
-using Elsa.Workflows.Runtime;
-using Elsa.Workflows.Runtime.Messages;
-using Microsoft.Extensions.DependencyInjection;
-
-namespace Elsa.Workflows.ComponentTests.Scenarios.BulkDispatchWorkflows;
-
-public class BulkDispatchWorkflowsTests : AppComponentTest
-{
-    private readonly SignalManager _signalManager;
-    private readonly IWorkflowRuntime _workflowRuntime;
-
-    public BulkDispatchWorkflowsTests(App app) : base(app)
-    {
-        _workflowRuntime = Scope.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
-        _signalManager = Scope.ServiceProvider.GetRequiredService<SignalManager>();
-    }
-
-    // /// <summary>
-    // /// Dispatches and waits for child workflows to complete.
-    // /// </summary>
-    // [Fact(Skip = "This test is flaky and needs to be fixed.")]
-    // public async Task DispatchAndWaitWorkflow_ShouldWaitForChildWorkflowToComplete()
-    // {
-    //     var workflowClient = await _workflowRuntime.CreateClientAsync();
-    //     await workflowClient.CreateInstanceAsync(new CreateWorkflowInstanceRequest
-    //     {
-    //         WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(GreetEmployeesWorkflow.DefinitionId, VersionOptions.Published)
-    //     });
-    //     await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty);
-    //     await _signalManager.WaitAsync<string>("Completed");
-    // }
-
-    /// <summary>
-    /// Individual items are sent as input to child workflows.
-    /// </summary>
-    [Fact]
-    public async Task DispatchWorkflows_ChildWorkflowsShouldReceiveCurrentItem()
-    {
-        var workflowClient = await _workflowRuntime.CreateClientAsync();
-        var request = new CreateAndRunWorkflowInstanceRequest
-        {
-            WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(MixFruitsWorkflow.DefinitionId, VersionOptions.Published)
-        };
-        await workflowClient.CreateAndRunInstanceAsync(request);
-
-        await _signalManager.WaitAsync("Apple");
-        await _signalManager.WaitAsync("Banana");
-        await _signalManager.WaitAsync("Cherry");
-    }
-}