Fix FlowJoin Activity NRE Bug (#5349)

* Update property retrieval in FlowJoin activity

This commit modifies the way flowScope is retrieved within the FlowJoin activity in Elsa.Workflows.Core. Instead of directly calling GetProperty, a fallback value is now being provided in case the desired property is not found. This reduces the risk of null reference exceptions.

* Add FlowJoins component tests

Two new files have been created to facilitate component testing for the FlowJoins scenarios in the Elsa Workflows. The `Tests.cs` file includes a Fact to validate the successful execution of a Flowchart with a single FlowJoin. The `Workflows.cs` file defines a single join workflow for these tests.

* Add RabbitMq support to ComponentTests

Added RabbitMq to Infrastructure.cs for component testing, allowing both RabbitMqContainer and DbContainer to start and stop asynchronously. Also, adjusted WorkflowServer.cs to configure RabbitMq mass transit alongside existing PostgreSql support, aiming to improve testing robustness and coverage.

* Refactor placement of RemoveOrphanedSubscriptions service

Move the implementation of AddNotificationHandler<RemoveOrphanedSubscriptions>() from the notifier block to the singleton section in AzureServiceBusFeature.cs. This will help keep all service registration related to notifications in one place.

* Increase prefetch count

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

@@ -323,7 +323,7 @@ services
                 {
                     massTransit.UseAzureServiceBus(azureServiceBusConnectionString, serviceBusFeature => serviceBusFeature.ConfigureServiceBus = bus =>
                     {
-                        bus.PrefetchCount = 4;
+                        bus.PrefetchCount = 100;
                         bus.LockDuration = TimeSpan.FromMinutes(5);
                         bus.MaxConcurrentCalls = 32;
                         bus.MaxDeliveryCount = 8;

src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs

@@ -108,7 +108,6 @@ public class AzureServiceBusFeature : FeatureBase
     {
         Services.Configure(AzureServiceBusOptions);
         Services.AddSingleton(ServiceBusAdministrationClientFactory);
-        Services.AddNotificationHandler<RemoveOrphanedSubscriptions>();
     }
 
     private static string GetConnectionString(IServiceProvider serviceProvider)
@@ -130,6 +129,7 @@ public class AzureServiceBusFeature : FeatureBase
         ).ToList();
 
         Services.AddSingleton(new MessageTopologyProvider(subscriptionTopology));
+        Services.AddNotificationHandler<RemoveOrphanedSubscriptions>();
     }
 
 }

src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs

@@ -38,7 +38,7 @@ public class FlowJoin : Activity, IJoinNode
         var flowchartContext = context.ParentActivityExecutionContext!;
         var flowchart = (Flowchart)flowchartContext.Activity;
         var inboundActivities = flowchart.Connections.LeftInboundActivities(this).ToList();
-        var flowScope = flowchartContext.GetProperty<FlowScope>(Flowchart.ScopeProperty)!;
+        var flowScope = flowchartContext.GetProperty(Flowchart.ScopeProperty, () => new FlowScope());
         var executionCount = flowScope.GetExecutionCount(this);
         var mode = context.Get(Mode);
 

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

@@ -1,4 +1,5 @@
 using Testcontainers.PostgreSql;
+using Testcontainers.RabbitMq;
 
 namespace Elsa.Workflows.ComponentTests;
 
@@ -11,13 +12,21 @@ public class Infrastructure : IAsyncLifetime
         .WithPassword("postgres")
         .Build();
 
+    public readonly RabbitMqContainer RabbitMqContainer = new RabbitMqBuilder()
+        .WithImage("rabbitmq:3-management")
+        .Build();
+
     public Task InitializeAsync()
     {
-        return DbContainer.StartAsync();
+        return Task.WhenAll(
+            DbContainer.StartAsync(),
+            RabbitMqContainer.StartAsync());
     }
 
     public Task DisposeAsync()
     {
-        return DbContainer.StopAsync();
+        return Task.WhenAll(
+            DbContainer.StopAsync(),
+            RabbitMqContainer.StopAsync());
     }
 }

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

@@ -3,6 +3,7 @@ using Elsa.EntityFrameworkCore.Extensions;
 using Elsa.EntityFrameworkCore.Modules.Management;
 using Elsa.Extensions;
 using Elsa.Identity.Providers;
+using Elsa.MassTransit.Extensions;
 using Elsa.Workflows.ComponentTests.Services;
 using FluentStorage;
 using Hangfire.Annotations;
@@ -37,6 +38,7 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl
     protected override void ConfigureWebHost(IWebHostBuilder builder)
     {
         var dbConnectionString = infrastructure.DbContainer.GetConnectionString();
+        var rabbitMqConnectionString = infrastructure.RabbitMqContainer.GetConnectionString();
 
         builder.UseUrls(url);
 
@@ -58,9 +60,18 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl
                     var workflowsDirectory = Path.Join(workflowsDirectorySegments);
                     return StorageFactory.Blobs.DirectoryFiles(workflowsDirectory);
                 });
+                elsa.UseMassTransit(massTransit =>
+                {
+                    massTransit.UseRabbitMq(rabbitMqConnectionString);
+                });
                 elsa.UseWorkflowManagement(management =>
                 {
                     management.UseEntityFrameworkCore(ef => ef.UsePostgreSql(dbConnectionString));
+                    management.UseMassTransitDispatcher();
+                });
+                elsa.UseWorkflowRuntime(runtime =>
+                {
+                    runtime.UseMassTransitDispatcher();
                 });
             };
         }

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

@@ -0,0 +1,16 @@
+using Elsa.Workflows.Contracts;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.FlowJoins;
+
+public class Tests(App app) : AppComponentTest(app)
+{
+    // https://github.com/elsa-workflows/elsa-core/issues/5348
+    [Fact]
+    public async Task FlowchartWithSingleFlowJoin_ShouldExecuteSuccessfully()
+    {
+        var workflowRunner = Scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
+        var result = await workflowRunner.RunAsync<SingleJoinWorkflow>();
+        Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
+    }
+}

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

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