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

Optimize workflow definition sync lookups (#7521)

Browse source 
* 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>
This commit is contained in:
Sipke Schoorstra 2026-05-21 02:42:51 +02:00 committed by GitHub
parent 2fa1a9ef8e
commit f7830f7063
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 77 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs
View file

@ -320,9 +320,8 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP
/// </summary>
private void SyncExistingCopies(List<WorkflowDefinition> primary, HashSet<WorkflowDefinition> 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);
}
}
}

61
test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs
View file

@ -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<WorkflowEvents>();
var completedChildWorkflows = new List<WorkflowState>();
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<List<WorkflowState>> WaitForCompletedChildWorkflowsAsync(string parentWorkflowInstanceId, string childWorkflowDefinitionId, int expectedChildCount)
{
var workflowInstanceStore = Scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
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<List<WorkflowState>> 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<IReadOnlyCollection<WorkflowInstance>> WaitForChildWorkflowInstancesAsync(

3
test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/Workflows/SlowBulkChildWorkflow.cs
View file

@ -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<string>(item)}")
}
};

65
test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/DispatchWorkflowsTests.cs
View file

@ -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<WorkflowExecutionContext> CompletedChildWorkflows)> RunWorkflowAndWaitForChildWorkflowAsync(
private async Task<(TestWorkflowExecutionResult Result, List<WorkflowState> CompletedChildWorkflows)> RunWorkflowAndWaitForChildWorkflowAsync(
string parentWorkflowDefinitionId,
string childWorkflowDefinitionId)
{
var workflowEvents = Scope.ServiceProvider.GetRequiredService<WorkflowEvents>();
var completedChildWorkflows = new List<WorkflowExecutionContext>();
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<List<WorkflowState>> WaitForCompletedChildWorkflowsAsync(string parentWorkflowInstanceId, string childWorkflowDefinitionId)
{
var workflowInstanceStore = Scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
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<List<WorkflowState>> 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();
}
}

5
test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/DispatchWorkflows/Workflows/SlowChildWorkflow.cs
View file

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