diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs index c982f49a2..cd874d358 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Activities/FlowJoin.cs @@ -33,10 +33,10 @@ public class FlowJoin : Activity, IJoinNode /// protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { - var flowchartExecutionContext = context.ParentActivityExecutionContext!; - var flowchart = (Flowchart)flowchartExecutionContext.Activity; + var flowchartContext = context.ParentActivityExecutionContext!; + var flowchart = (Flowchart)flowchartContext.Activity; var inboundActivities = flowchart.Connections.LeftInboundActivities(this).ToList(); - var flowScope = flowchartExecutionContext.GetProperty(Flowchart.ScopeProperty)!; + var flowScope = flowchartContext.GetProperty(Flowchart.ScopeProperty)!; var executionCount = flowScope.GetExecutionCount(this); var mode = context.Get(Mode); @@ -49,7 +49,7 @@ public class FlowJoin : Activity, IJoinNode if (haveAllInboundActivitiesExecuted) { - await ClearBookmarksAsync(flowchart, context); + await CancelActivitiesInInboundPathAsync(flowchart, flowchartContext, context); await context.CompleteActivityAsync(); } @@ -57,20 +57,20 @@ public class FlowJoin : Activity, IJoinNode } case FlowJoinMode.WaitAny: { - await ClearBookmarksAsync(flowchart, context); + await CancelActivitiesInInboundPathAsync(flowchart, flowchartContext, context); await context.CompleteActivityAsync(); break; } } } - private async Task ClearBookmarksAsync(Flowchart flowchart, ActivityExecutionContext context) + private async Task CancelActivitiesInInboundPathAsync(Flowchart flowchart, ActivityExecutionContext flowchartContext, ActivityExecutionContext joinContext) { // Cancel all activities between this join activity and its most recent fork. var connections = flowchart.Connections; - var workflowExecutionContext = context.WorkflowExecutionContext; + var workflowExecutionContext = joinContext.WorkflowExecutionContext; var inboundActivities = connections.LeftAncestorActivities(this).Select(x => workflowExecutionContext.FindNodeByActivity(x)).Select(x => x!.Activity).ToList(); - var inboundActivityExecutionContexts = workflowExecutionContext.ActivityExecutionContexts.Where(x => inboundActivities.Contains(x.Activity)).ToList(); + var inboundActivityExecutionContexts = workflowExecutionContext.ActivityExecutionContexts.Where(x => inboundActivities.Contains(x.Activity) && x.ParentActivityExecutionContext == flowchartContext).ToList(); // Cancel each inbound activity. foreach (var activityExecutionContext in inboundActivityExecutionContexts) diff --git a/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs index 3f91e92f7..ee24f57cd 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/WorkflowExecutionContextExtensions.cs @@ -98,7 +98,8 @@ public static class WorkflowExecutionContextExtensions if (options?.PreventDuplicateScheduling == true) { - var existingWorkItem = scheduler.Find(x => x.Activity.NodeId == activityNode.NodeId); + // Check if the activity is already scheduled for the specified owner. + var existingWorkItem = scheduler.Find(x => x.Activity.NodeId == activityNode.NodeId && x.Owner == owner); if (existingWorkItem != null) return existingWorkItem; diff --git a/test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj b/test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj index e171adab3..29f7310ed 100644 --- a/test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj +++ b/test/integration/Elsa.IntegrationTests/Elsa.IntegrationTests.csproj @@ -153,5 +153,11 @@ Always + + Always + + + Always + diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/JoinRunsOnceTests.cs b/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/JoinRunsOnceTests.cs new file mode 100644 index 000000000..d1e9c5a84 --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/JoinRunsOnceTests.cs @@ -0,0 +1,52 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using Elsa.Common.Models; +using Elsa.IntegrationTests.Scenarios.ImplicitJoins.Workflows; +using Elsa.Testing.Shared; +using Elsa.Workflows.Core; +using Elsa.Workflows.Core.Contracts; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Filters; +using Microsoft.Extensions.DependencyInjection; +using Open.Linq.AsyncExtensions; +using Xunit; +using Xunit.Abstractions; + +namespace Elsa.IntegrationTests.Scenarios.ImplicitJoins; + +public class JoinRunsOnceTests +{ + private readonly IWorkflowRunner _workflowRunner; + private readonly CapturingTextWriter _capturingTextWriter = new(); + private readonly IServiceProvider _services; + + public JoinRunsOnceTests(ITestOutputHelper testOutputHelper) + { + _services = new TestApplicationBuilder(testOutputHelper).WithCapturingTextWriter(_capturingTextWriter).Build(); + _workflowRunner = _services.GetRequiredService(); + } + + [Fact(DisplayName = "The Join activity executes only once, not twice")] + public async Task Test1() + { + // Populate registries. + await _services.PopulateRegistriesAsync(); + + // Import workflow. + var workflowDefinition = await _services.ImportWorkflowDefinitionAsync($"Scenarios/ImplicitJoins/Workflows/join.json"); + + // Execute. + var state = await _services.RunWorkflowUntilEndAsync(workflowDefinition.DefinitionId); + + // Assert. + var journal = await _services.GetRequiredService().FindManyAsync(new WorkflowExecutionLogRecordFilter + { + WorkflowInstanceId = state.Id, + ActivityId = "802725996be1b582", + EventName = "Completed" + }, PageArgs.All); + + Assert.Single(journal.Items); + } +} \ No newline at end of file diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/ParallelJoinCompletesTests.cs b/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/ParallelJoinCompletesTests.cs new file mode 100644 index 000000000..3941705cb --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/ParallelJoinCompletesTests.cs @@ -0,0 +1,52 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using Elsa.Common.Models; +using Elsa.IntegrationTests.Scenarios.ImplicitJoins.Workflows; +using Elsa.Testing.Shared; +using Elsa.Workflows.Core; +using Elsa.Workflows.Core.Contracts; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Filters; +using Microsoft.Extensions.DependencyInjection; +using Open.Linq.AsyncExtensions; +using Xunit; +using Xunit.Abstractions; + +namespace Elsa.IntegrationTests.Scenarios.ImplicitJoins; + +public class ParallelJoinCompletesTests +{ + private readonly IWorkflowRunner _workflowRunner; + private readonly CapturingTextWriter _capturingTextWriter = new(); + private readonly IServiceProvider _services; + + public ParallelJoinCompletesTests(ITestOutputHelper testOutputHelper) + { + _services = new TestApplicationBuilder(testOutputHelper).WithCapturingTextWriter(_capturingTextWriter).Build(); + _workflowRunner = _services.GetRequiredService(); + } + + [Fact(DisplayName = "The ParallelForEach activity completes when its Body contains a Join activity")] + public async Task Test1() + { + // Populate registries. + await _services.PopulateRegistriesAsync(); + + // Import workflow. + var workflowDefinition = await _services.ImportWorkflowDefinitionAsync($"Scenarios/ImplicitJoins/Workflows/parallel-join.json"); + + // Execute. + var state = await _services.RunWorkflowUntilEndAsync(workflowDefinition.DefinitionId); + + // Assert. + var journal = await _services.GetRequiredService().FindManyAsync(new WorkflowExecutionLogRecordFilter + { + WorkflowInstanceId = state.Id, + ActivityId = "70fc1183cd5800f2", + EventName = "Completed" + }, PageArgs.All); + + Assert.Single(journal.Items); + } +} \ No newline at end of file diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/Workflows/join.json b/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/Workflows/join.json new file mode 100644 index 000000000..dc3fad71e --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/Workflows/join.json @@ -0,0 +1,215 @@ +{ + "id": "825e9d347e2f3094", + "definitionId": "f64e42304661f407", + "name": "Joins", + "createdAt": "2023-12-13T21:41:53.628415+00:00", + "version": 3, + "toolVersion": "3.0.0.0", + "variables": [], + "inputs": [], + "outputs": [], + "outcomes": [], + "customProperties": { + "Elsa:WorkflowContextProviderTypes": [] + }, + "isReadonly": false, + "isLatest": true, + "isPublished": false, + "options": { + "autoUpdateConsumingWorkflows": false + }, + "root": { + "type": "Elsa.Flowchart", + "version": 1, + "id": "8a03bf79d15fd48a", + "nodeId": "Workflow1:8a03bf79d15fd48a", + "metadata": {}, + "customProperties": { + "source": "FlowchartJsonConverter.cs:45", + "notFoundConnections": [], + "canStartWorkflow": false, + "runAsynchronously": false + }, + "activities": [ + { + "id": "c7624645ff77c7d0", + "nodeId": "Workflow1:8a03bf79d15fd48a:c7624645ff77c7d0", + "name": "Start1", + "type": "Elsa.Start", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": -400.5, + "y": 212 + }, + "size": { + "width": 102.21875, + "height": 50 + } + } + } + }, + { + "text": null, + "id": "281e9ea6d278b231", + "nodeId": "Workflow1:8a03bf79d15fd48a:281e9ea6d278b231", + "name": "WriteLine2", + "type": "Elsa.WriteLine", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": -160.5, + "y": 305 + }, + "size": { + "width": 139.296875, + "height": 50 + } + } + } + }, + { + "mode": { + "typeName": "Elsa.Workflows.Core.Activities.Flowchart.Models.FlowJoinMode, Elsa.Workflows.Core", + "expression": { + "type": "Literal", + "value": "WaitAny" + }, + "memoryReference": { + "id": "802725996be1b582:input-0" + } + }, + "id": "802725996be1b582", + "nodeId": "Workflow1:8a03bf79d15fd48a:802725996be1b582", + "name": "FlowJoin1", + "type": "Elsa.FlowJoin", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": 80, + "y": 200 + }, + "size": { + "width": 98.265625, + "height": 50 + } + } + } + }, + { + "text": null, + "id": "bfc88f064cf394ed", + "nodeId": "Workflow1:8a03bf79d15fd48a:bfc88f064cf394ed", + "name": "WriteLine1", + "type": "Elsa.WriteLine", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": -160, + "y": 120 + }, + "size": { + "width": 139.296875, + "height": 50 + } + } + } + }, + { + "id": "5467123c7973d011", + "nodeId": "Workflow1:8a03bf79d15fd48a:5467123c7973d011", + "name": "End1", + "type": "Elsa.End", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": 274.5, + "y": 200 + }, + "size": { + "width": 94.40625, + "height": 50 + } + } + } + } + ], + "connections": [ + { + "source": { + "activity": "c7624645ff77c7d0", + "port": "Done" + }, + "target": { + "activity": "bfc88f064cf394ed", + "port": "In" + } + }, + { + "source": { + "activity": "bfc88f064cf394ed", + "port": "Done" + }, + "target": { + "activity": "802725996be1b582", + "port": "In" + } + }, + { + "source": { + "activity": "c7624645ff77c7d0", + "port": "Done" + }, + "target": { + "activity": "281e9ea6d278b231", + "port": "In" + } + }, + { + "source": { + "activity": "281e9ea6d278b231", + "port": "Done" + }, + "target": { + "activity": "802725996be1b582", + "port": "In" + } + }, + { + "source": { + "activity": "802725996be1b582", + "port": "Done" + }, + "target": { + "activity": "5467123c7973d011", + "port": "In" + } + } + ] + } +} \ No newline at end of file diff --git a/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/Workflows/parallel-join.json b/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/Workflows/parallel-join.json new file mode 100644 index 000000000..2672924d1 --- /dev/null +++ b/test/integration/Elsa.IntegrationTests/Scenarios/ImplicitJoins/Workflows/parallel-join.json @@ -0,0 +1,241 @@ +{ + "id": "2effd8f5f4205a5b", + "definitionId": "2da8d4cb0fcfbacd", + "name": "Parallel \u002B Join", + "createdAt": "2023-12-13T21:42:12.634042+00:00", + "version": 2, + "toolVersion": "3.0.0.0", + "variables": [], + "inputs": [], + "outputs": [], + "outcomes": [], + "customProperties": { + "Elsa:WorkflowContextProviderTypes": [] + }, + "isReadonly": false, + "isLatest": true, + "isPublished": false, + "options": { + "autoUpdateConsumingWorkflows": false + }, + "root": { + "type": "Elsa.Flowchart", + "version": 1, + "id": "d3350290ed96e720", + "nodeId": "Workflow1:d3350290ed96e720", + "metadata": {}, + "customProperties": { + "source": "FlowchartJsonConverter.cs:45", + "notFoundConnections": [], + "canStartWorkflow": false, + "runAsynchronously": false + }, + "activities": [ + { + "items": { + "typeName": "Object[]", + "expression": { + "type": "JavaScript", + "value": "[1,2,3]" + }, + "memoryReference": { + "id": "70fc1183cd5800f2:input-0" + } + }, + "body": { + "type": "Elsa.Flowchart", + "version": 1, + "id": "b07170f2d68980cd", + "nodeId": "Workflow1:d3350290ed96e720:70fc1183cd5800f2:b07170f2d68980cd", + "metadata": {}, + "customProperties": { + "source": "FlowchartJsonConverter.cs:45", + "notFoundConnections": [], + "canStartWorkflow": false, + "runAsynchronously": false + }, + "activities": [ + { + "condition": { + "typeName": "Boolean", + "expression": { + "type": "JavaScript", + "value": "false" + }, + "memoryReference": { + "id": "33665f061d96b957:input-0" + } + }, + "id": "33665f061d96b957", + "nodeId": "Workflow1:d3350290ed96e720:70fc1183cd5800f2:b07170f2d68980cd:33665f061d96b957", + "name": "FlowDecision1", + "type": "Elsa.FlowDecision", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": -329.140625, + "y": -78.5 + }, + "size": { + "width": 129.515625, + "height": 50 + } + } + } + }, + { + "text": null, + "id": "e27177eb5965bd6b", + "nodeId": "Workflow1:d3350290ed96e720:70fc1183cd5800f2:b07170f2d68980cd:e27177eb5965bd6b", + "name": "WriteLine1", + "type": "Elsa.WriteLine", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": -40, + "y": -160 + }, + "size": { + "width": 139.296875, + "height": 50 + } + } + } + }, + { + "text": null, + "id": "343d233a75d8d8e9", + "nodeId": "Workflow1:d3350290ed96e720:70fc1183cd5800f2:b07170f2d68980cd:343d233a75d8d8e9", + "name": "WriteLine2", + "type": "Elsa.WriteLine", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": 371.859375, + "y": -78.5 + }, + "size": { + "width": 139.296875, + "height": 50 + } + } + } + }, + { + "mode": { + "typeName": "Elsa.Workflows.Core.Activities.Flowchart.Models.FlowJoinMode, Elsa.Workflows.Core", + "expression": { + "type": "Literal", + "value": "WaitAny" + }, + "memoryReference": { + "id": "1f8625e278b74aa3:input-0" + } + }, + "id": "1f8625e278b74aa3", + "nodeId": "Workflow1:d3350290ed96e720:70fc1183cd5800f2:b07170f2d68980cd:1f8625e278b74aa3", + "name": "FlowJoin1", + "type": "Elsa.FlowJoin", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": 188.859375, + "y": -78.5 + }, + "size": { + "width": 98.265625, + "height": 50 + } + } + } + } + ], + "connections": [ + { + "source": { + "activity": "33665f061d96b957", + "port": "True" + }, + "target": { + "activity": "e27177eb5965bd6b", + "port": "In" + } + }, + { + "source": { + "activity": "e27177eb5965bd6b", + "port": "Done" + }, + "target": { + "activity": "1f8625e278b74aa3", + "port": "In" + } + }, + { + "source": { + "activity": "1f8625e278b74aa3", + "port": "Done" + }, + "target": { + "activity": "343d233a75d8d8e9", + "port": "In" + } + }, + { + "source": { + "activity": "33665f061d96b957", + "port": "False" + }, + "target": { + "activity": "1f8625e278b74aa3", + "port": "In" + } + } + ] + }, + "id": "70fc1183cd5800f2", + "nodeId": "Workflow1:d3350290ed96e720:70fc1183cd5800f2", + "name": "ParallelForEach1", + "type": "Elsa.ParallelForEach", + "version": 1, + "customProperties": { + "canStartWorkflow": false, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": -260, + "y": 140 + }, + "size": { + "width": 185.203125, + "height": 116.015625 + } + } + } + } + ], + "connections": [] + } +} \ No newline at end of file