From f7830f706326bbeab5c371c1a8bac51a03631111 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Thu, 21 May 2026 02:42:51 +0200 Subject: [PATCH] Optimize workflow definition sync lookups (#7521) * Optimize workflow definition sync lookups Replace repeated linear scans in workflow definition synchronization with a single hash set lookup. This keeps the existing replacement behavior while avoiding quadratic work as the number of workflow definitions grows. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Stabilize bulk dispatch component test Wait for child workflow completion through the workflow instance store instead of relying on in-memory event delivery, and avoid using the scheduled Delay activity for the fire-and-forget child workflow timing check. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Stabilize dispatch component test Use persisted workflow instance state when waiting for fire-and-forget child workflow completion, and avoid scheduled Delay for the timing check. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Add missing using for merge conflict resolution Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Remove unused bulk dispatch using Drop an unnecessary using from the component test file so the PR branch has a new maintainer-authored head commit for CI. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --- ...DefaultWorkflowDefinitionStorePopulator.cs | 9 ++- .../BulkDispatchWorkflowsTests.cs | 61 +++++++++-------- .../Workflows/SlowBulkChildWorkflow.cs | 3 +- .../DispatchWorkflowsTests.cs | 65 +++++++++++-------- .../Workflows/SlowChildWorkflow.cs | 5 +- 5 files changed, 77 insertions(+), 66 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs index 092343833..298c9fab3 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs @@ -320,9 +320,8 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP /// private void SyncExistingCopies(List primary, HashSet secondary) { - var ids = secondary.Select(x => x.Id).Distinct().ToList(); - var latestWorkflowDefinitions = primary.Where(x => ids.Contains(x.Id)).ToList(); - primary.RemoveAll(x => latestWorkflowDefinitions.Contains(x)); - primary.AddRange(secondary.Where(x => ids.Contains(x.Id))); + var ids = secondary.Select(x => x.Id).ToHashSet(); + primary.RemoveAll(x => ids.Contains(x.Id)); + primary.AddRange(secondary); } -} \ No newline at end of file +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs index 5f4f636b3..847bdebee 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs @@ -1,6 +1,5 @@ using System.Diagnostics; using Elsa.Common.Models; -using Elsa.Testing.Shared; using Elsa.Testing.Shared.Models; using Elsa.Testing.Shared.Services; using Elsa.Workflows.Activities; @@ -148,40 +147,46 @@ public class BulkDispatchWorkflowsTests : AppComponentTest string childWorkflowDefinitionId, int expectedChildCount) { - var workflowEvents = Scope.ServiceProvider.GetRequiredService(); - var completedChildWorkflows = new List(); - var childWorkflowCompletionTcs = new TaskCompletionSource(); + var result = await RunWorkflowAsync(parentWorkflowDefinitionId); + var completedChildWorkflows = await WaitForCompletedChildWorkflowsAsync(result.WorkflowExecutionContext.Id, childWorkflowDefinitionId, expectedChildCount); - // Subscribe to child workflow completion events - void OnWorkflowStateCommitted(object? sender, WorkflowStateCommittedEventArgs e) + return (result, completedChildWorkflows); + } + + private async Task> WaitForCompletedChildWorkflowsAsync(string parentWorkflowInstanceId, string childWorkflowDefinitionId, int expectedChildCount) + { + var workflowInstanceStore = Scope.ServiceProvider.GetRequiredService(); + var timeoutAt = DateTimeOffset.UtcNow.AddSeconds(ChildWorkflowTimeoutSeconds); + + while (DateTimeOffset.UtcNow < timeoutAt) { - if (e.WorkflowState.DefinitionId != childWorkflowDefinitionId || - e.WorkflowState.Status != WorkflowStatus.Finished) - { - return; - } + var completedChildWorkflows = await FindChildWorkflowStatesAsync(workflowInstanceStore, parentWorkflowInstanceId, childWorkflowDefinitionId, WorkflowStatus.Finished); - completedChildWorkflows.Add(e.WorkflowState); - if (completedChildWorkflows.Count == expectedChildCount) - childWorkflowCompletionTcs.TrySetResult(); + if (completedChildWorkflows.Count >= expectedChildCount) + return completedChildWorkflows; + + await Task.Delay(TimeSpan.FromMilliseconds(100)); } - workflowEvents.WorkflowStateCommitted += OnWorkflowStateCommitted; + var childWorkflows = await FindChildWorkflowStatesAsync(workflowInstanceStore, parentWorkflowInstanceId, childWorkflowDefinitionId); + var observedStates = string.Join(", ", childWorkflows.Select(x => $"{x.Id}:{x.Status}:{x.SubStatus}")); + throw new TimeoutException($"Expected {expectedChildCount} completed child workflows with definition ID {childWorkflowDefinitionId}, but observed {childWorkflows.Count(x => x.Status == WorkflowStatus.Finished)}. Observed child workflow states: {observedStates}"); + } - try + private static async Task> FindChildWorkflowStatesAsync( + IWorkflowInstanceStore workflowInstanceStore, + string parentWorkflowInstanceId, + string childWorkflowDefinitionId, + WorkflowStatus? workflowStatus = null) + { + var filter = new WorkflowInstanceFilter { - // Run the main workflow - var result = await RunWorkflowAsync(parentWorkflowDefinitionId); - - // Wait for all child workflows to complete - await childWorkflowCompletionTcs.Task.WaitAsync(TimeSpan.FromSeconds(ChildWorkflowTimeoutSeconds)); - - return (result, completedChildWorkflows); - } - finally - { - workflowEvents.WorkflowStateCommitted -= OnWorkflowStateCommitted; - } + DefinitionId = childWorkflowDefinitionId, + ParentWorkflowInstanceIds = new[] { parentWorkflowInstanceId }, + WorkflowStatus = workflowStatus + }; + var instances = await workflowInstanceStore.FindManyAsync(filter); + return instances.Select(x => x.WorkflowState).ToList(); } private async Task> WaitForChildWorkflowInstancesAsync( diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/Workflows/SlowBulkChildWorkflow.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/Workflows/SlowBulkChildWorkflow.cs index 2809071a0..d004b4628 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/Workflows/SlowBulkChildWorkflow.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/Workflows/SlowBulkChildWorkflow.cs @@ -1,5 +1,4 @@ using Elsa.Extensions; -using Elsa.Scheduling.Activities; using Elsa.Workflows.Activities; using JetBrains.Annotations; @@ -19,7 +18,7 @@ public class SlowBulkChildWorkflow : WorkflowBase { Activities = { - Delay.FromMilliseconds(10), + new Inline(context => new ValueTask(Task.Delay(10, context.CancellationToken))), new WriteLine(context => $"Processing item: {context.GetInput(item)}") } }; diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/DispatchWorkflowsTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/DispatchWorkflowsTests.cs index 938b01170..3caff638e 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/DispatchWorkflowsTests.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/DispatchWorkflowsTests.cs @@ -1,5 +1,4 @@ using Elsa.Common.Models; -using Elsa.Testing.Shared; using Elsa.Testing.Shared.Models; using Elsa.Testing.Shared.Services; using Elsa.Workflows.Activities; @@ -7,7 +6,10 @@ using Elsa.Workflows.ComponentTests.Abstractions; using Elsa.Workflows.ComponentTests.Fixtures; using Elsa.Workflows.ComponentTests.Scenarios.Activities.Composition.DispatchWorkflows.Workflows; using Elsa.Workflows.ComponentTests.Scenarios.Activities.DispatchWorkflows.Workflows; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Models; +using Elsa.Workflows.State; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.Composition.DispatchWorkflows; @@ -99,42 +101,49 @@ public class DispatchWorkflowsTests : AppComponentTest Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowExecutionContext.SubStatus); } - private async Task<(TestWorkflowExecutionResult Result, List CompletedChildWorkflows)> RunWorkflowAndWaitForChildWorkflowAsync( + private async Task<(TestWorkflowExecutionResult Result, List CompletedChildWorkflows)> RunWorkflowAndWaitForChildWorkflowAsync( string parentWorkflowDefinitionId, string childWorkflowDefinitionId) { - var workflowEvents = Scope.ServiceProvider.GetRequiredService(); - var completedChildWorkflows = new List(); - var childWorkflowCompletionTcs = new TaskCompletionSource(); + var result = await RunWorkflowAsync(parentWorkflowDefinitionId); + var completedChildWorkflows = await WaitForCompletedChildWorkflowsAsync(result.WorkflowExecutionContext.Id, childWorkflowDefinitionId); - // Subscribe to child workflow completion events - void OnWorkflowStateCommitted(object? sender, WorkflowStateCommittedEventArgs e) + return (result, completedChildWorkflows); + } + + private async Task> WaitForCompletedChildWorkflowsAsync(string parentWorkflowInstanceId, string childWorkflowDefinitionId) + { + var workflowInstanceStore = Scope.ServiceProvider.GetRequiredService(); + var timeoutAt = DateTimeOffset.UtcNow.AddSeconds(ChildWorkflowTimeoutSeconds); + + while (DateTimeOffset.UtcNow < timeoutAt) { - if (e.WorkflowExecutionContext.Workflow.Identity.DefinitionId != childWorkflowDefinitionId || - e.WorkflowExecutionContext.Status != WorkflowStatus.Finished) - { - return; - } + var completedChildWorkflows = await FindChildWorkflowStatesAsync(workflowInstanceStore, parentWorkflowInstanceId, childWorkflowDefinitionId, WorkflowStatus.Finished); - completedChildWorkflows.Add(e.WorkflowExecutionContext); - childWorkflowCompletionTcs.TrySetResult(); + if (completedChildWorkflows.Count > 0) + return completedChildWorkflows; + + await Task.Delay(TimeSpan.FromMilliseconds(100)); } - workflowEvents.WorkflowStateCommitted += OnWorkflowStateCommitted; + var childWorkflows = await FindChildWorkflowStatesAsync(workflowInstanceStore, parentWorkflowInstanceId, childWorkflowDefinitionId); + var observedStates = string.Join(", ", childWorkflows.Select(x => $"{x.Id}:{x.Status}:{x.SubStatus}")); + throw new TimeoutException($"Expected a completed child workflow with definition ID {childWorkflowDefinitionId}, but observed none. Observed child workflow states: {observedStates}"); + } - try + private static async Task> FindChildWorkflowStatesAsync( + IWorkflowInstanceStore workflowInstanceStore, + string parentWorkflowInstanceId, + string childWorkflowDefinitionId, + WorkflowStatus? workflowStatus = null) + { + var filter = new WorkflowInstanceFilter { - // Run the main workflow - var result = await RunWorkflowAsync(parentWorkflowDefinitionId); - - // Wait for the child workflow to complete - await childWorkflowCompletionTcs.Task.WaitAsync(TimeSpan.FromSeconds(ChildWorkflowTimeoutSeconds)); - - return (result, completedChildWorkflows); - } - finally - { - workflowEvents.WorkflowStateCommitted -= OnWorkflowStateCommitted; - } + DefinitionId = childWorkflowDefinitionId, + ParentWorkflowInstanceIds = new[] { parentWorkflowInstanceId }, + WorkflowStatus = workflowStatus + }; + var instances = await workflowInstanceStore.FindManyAsync(filter); + return instances.Select(x => x.WorkflowState).ToList(); } } diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/Workflows/SlowChildWorkflow.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/Workflows/SlowChildWorkflow.cs index bfacd28db..e8348c732 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/Workflows/SlowChildWorkflow.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/Workflows/SlowChildWorkflow.cs @@ -1,4 +1,3 @@ -using Elsa.Scheduling.Activities; using Elsa.Workflows.Activities; using JetBrains.Annotations; @@ -16,9 +15,9 @@ public class SlowChildWorkflow : WorkflowBase { Activities = { - Delay.FromMilliseconds(10), + new Inline(context => new ValueTask(Task.Delay(10, context.CancellationToken))), new WriteLine("Slow child workflow executed") } }; } -} \ No newline at end of file +}