Addresses workflow identity reload conflict (#6909)

* Refactor `IndexTriggersAsync` to use `WorkflowDefinition` and update null assignment for serialization logic.

* Change default parameter value from `default` to `null` in `UseFluentStorageProvider` method signature.

* Add in-memory workflows provider and materializer for integration tests

Introduced `InMemoryWorkflowsProvider` and `InMemoryWorkflowMaterializer` to support integration testing scenarios for workflow definition population. Enhanced workflow handling with fluent method `WithId` for `WorkflowBuilder`. Updated event publishing and dependency injection logic.

* Remove extraneous whitespace in DefaultWorkflowDefinitionStorePopulator.

* Refine test class documentation for `WorkflowDefinitionStorePopulation` scenario.
This commit is contained in:
Sipke Schoorstra 2025-09-15 11:56:01 +02:00 committed by GitHub
parent c019cfa3da
commit b613ff6b61
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 148 additions and 13 deletions

View file

@ -29,7 +29,7 @@ public static class ModuleExtensions
/// <param name="module">The module.</param>
/// <param name="configure">The configuration delegate.</param>
/// <returns>The module.</returns>
public static IModule UseFluentStorageProvider(this IModule module, Action<BlobStorageFeature>? configure = default)
public static IModule UseFluentStorageProvider(this IModule module, Action<BlobStorageFeature>? configure = null)
{
module.Use(configure);
return module;

View file

@ -63,6 +63,12 @@ public class WorkflowBuilder(IActivityVisitor activityVisitor, IIdentityGraphSer
DefinitionId = definitionId;
return this;
}
public IWorkflowBuilder WithId(string id)
{
Id = id;
return this;
}
/// <inheritdoc />
public IWorkflowBuilder WithTenantId(string tenantId)

View file

@ -96,6 +96,13 @@ public interface IWorkflowBuilder
/// <param name="definitionId">The definition ID to use for the workflow being built.</param>
IWorkflowBuilder WithDefinitionId(string definitionId);
/// <summary>
/// A fluent method for setting the <see cref="Id"/> property.
/// </summary>
/// <param name="id">The unique identifier to use for the workflow being built.</param>
/// <returns>The current <see cref="IWorkflowBuilder"/> instance for method chaining.</returns>
IWorkflowBuilder WithId(string id);
/// <summary>
/// A fluent method for setting the <see cref="TenantId"/> property.
/// </summary>

View file

@ -15,7 +15,7 @@ namespace Elsa.Workflows.Runtime.Activities;
[UsedImplicitly]
public class Event : Trigger<object?>
{
internal const string EventInputWorkflowInputKey = "__EventPayloadWorkflowInput";
public const string EventInputWorkflowInputKey = "__EventPayloadWorkflowInput";
/// <inheritdoc />
internal Event([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)

View file

@ -10,10 +10,15 @@ namespace Microsoft.Extensions.DependencyInjection;
public static class DependencyInjectionExtensions
{
/// <summary>
/// Adds the <see cref="ClrWorkflowsProvider"/> to the service collection.
/// Adds the specified workflow provider type to the service collection.
/// </summary>
/// <param name="services">The service collection.</param>
/// <typeparam name="T">The type of the workflow definition provider.</typeparam>
/// <returns>The service collection.</returns>
/// <typeparam name="T">The type of the workflow provider to add. Must implement <see cref="IWorkflowsProvider"/>.</typeparam>
[Obsolete("Use AddWorkflowsProvider instead.", false)]
public static IServiceCollection AddWorkflowDefinitionProvider<T>(this IServiceCollection services) where T : class, IWorkflowsProvider => services.AddScoped<IWorkflowsProvider, T>();
/// <summary>
/// Adds the specified workflows provider type to the service collection.
/// </summary>
/// <typeparam name="T">The type of the workflow provider to add. Must implement <see cref="IWorkflowsProvider"/>.</typeparam>
public static IServiceCollection AddWorkflowsProvider<T>(this IServiceCollection services) where T : class, IWorkflowsProvider => services.AddScoped<IWorkflowsProvider, T>();
}

View file

@ -9,4 +9,4 @@ namespace Elsa.Workflows.Runtime;
/// <param name="ProviderName">The name of the provider that provided the workflow definition.</param>
/// <param name="MaterializerName">The name of the materializer that materialized the workflow.</param>
/// <param name="MaterializerContext">The context of the materializer that materialized the workflow.</param>
public record MaterializedWorkflow(Workflow Workflow, string ProviderName, string MaterializerName, object? MaterializerContext = default);
public record MaterializedWorkflow(Workflow Workflow, string ProviderName, string MaterializerName, object? MaterializerContext = null);

View file

@ -56,7 +56,7 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP
{
var providers = _workflowDefinitionProviders();
var workflowDefinitions = new List<WorkflowDefinition>();
foreach (var provider in providers)
{
var results = await provider.GetWorkflowsAsync(cancellationToken).AsTask().ToList();
@ -84,7 +84,7 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP
var workflowDefinition = await AddOrUpdateAsync(materializedWorkflow, cancellationToken);
if (indexTriggers)
await IndexTriggersAsync(materializedWorkflow, cancellationToken);
await IndexTriggersAsync(workflowDefinition, cancellationToken);
return workflowDefinition;
}
@ -118,7 +118,7 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP
// Serialize materializer context.
var materializerContext = materializedWorkflow.MaterializerContext;
var materializerContextJson = materializerContext != null ? _payloadSerializer.Serialize(materializerContext) : default;
var materializerContextJson = materializerContext != null ? _payloadSerializer.Serialize(materializerContext) : null;
// Serialize the workflow root.
var workflowJson = _activitySerializer.Serialize(workflow.Root);
@ -260,7 +260,7 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP
}
}
private async Task IndexTriggersAsync(MaterializedWorkflow materializedWorkflow, CancellationToken cancellationToken) => await _triggerIndexer.IndexTriggersAsync(materializedWorkflow.Workflow, cancellationToken);
private async Task IndexTriggersAsync(WorkflowDefinition workflowDefinition, CancellationToken cancellationToken) => await _triggerIndexer.IndexTriggersAsync(workflowDefinition, cancellationToken);
/// <summary>
/// Syncs the items in the primary list with existing items in the secondary list, even when the object instances are not the same (but their IDs are).

View file

@ -29,16 +29,17 @@ public class EventPublisher(IStimulusSender stimulusSender, IStimulusDispatcher
WorkflowInstanceId = workflowInstanceId,
Input = workflowInput
};
var triggerName = ActivityTypeNameHelper.GenerateTypeName<Event>();
if (asynchronous)
{
await stimulusDispatcher.SendAsync(new()
{
ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<Event>(),
ActivityTypeName = triggerName,
Stimulus = stimulus,
Metadata = metadata
}, cancellationToken);
}
else
await stimulusSender.SendAsync<Event>(stimulus, metadata, cancellationToken);
await stimulusSender.SendAsync(triggerName, stimulus, metadata, cancellationToken);
}
}

View file

@ -0,0 +1,20 @@
using Elsa.Workflows.Activities;
using Elsa.Workflows.Management;
using Elsa.Workflows.Management.Entities;
namespace Elsa.Workflows.IntegrationTests.Scenarios.WorkflowDefinitionStorePopulation;
public class InMemoryWorkflowMaterializer(Workflow workflow) : IWorkflowMaterializer
{
public string Name => "InMemory";
public ValueTask<Workflow> MaterializeAsync(WorkflowDefinition definition, CancellationToken cancellationToken = default)
{
var materializedWorkflow = new Workflow
{
Identity = new(definition.DefinitionId, definition.Version, definition.Id),
Root = workflow.Root
};
return new(materializedWorkflow);
}
}

View file

@ -0,0 +1,23 @@
using Elsa.Workflows.Activities;
using Elsa.Workflows.Models;
using Elsa.Workflows.Runtime;
using Elsa.Workflows.Runtime.Activities;
namespace Elsa.Workflows.IntegrationTests.Scenarios.WorkflowDefinitionStorePopulation;
public class InMemoryWorkflowsProvider(Workflow workflow) : IWorkflowsProvider
{
public string Name => "InMemory";
public ValueTask<IEnumerable<MaterializedWorkflow>> GetWorkflowsAsync(CancellationToken cancellationToken = default)
{
var materializedWorkflow = new MaterializedWorkflow(
Workflow: workflow,
ProviderName: "InMemory",
MaterializerName: "InMemory",
MaterializerContext: null
);
return new([materializedWorkflow]);
}
}

View file

@ -0,0 +1,73 @@
using Elsa.Testing.Shared;
using Elsa.Workflows.Activities;
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Management;
using Elsa.Workflows.Runtime;
using Elsa.Workflows.Runtime.Activities;
using Elsa.Workflows.Runtime.Stimuli;
using Microsoft.Extensions.DependencyInjection;
using Xunit.Abstractions;
namespace Elsa.Workflows.IntegrationTests.Scenarios.WorkflowDefinitionStorePopulation;
/// <summary>
/// Represents a test class for integration testing various scenarios related to the
/// `WorkflowDefinitionStorePopulation`. This class primarily focuses
/// on ensuring correct behavior when workflow definitions are published or updated, and their effects
/// on consuming workflows and triggers.
/// </summary>
public class Tests
{
private readonly IServiceProvider _services;
private readonly Workflow _shiftyWorkflow;
public Tests(ITestOutputHelper testOutputHelper)
{
_shiftyWorkflow = new()
{
Identity = new(
DefinitionId: "WorkflowWithTrigger",
Version: 1,
Id: "1",
TenantId: "default"
),
Root = new Event("Foo")
{
CanStartWorkflow = true
}
};
_services = new TestApplicationBuilder(testOutputHelper)
.ConfigureServices(services => services
.AddScoped<IWorkflowsProvider>(_ => new InMemoryWorkflowsProvider(_shiftyWorkflow))
.AddScoped<IWorkflowMaterializer>(_ => new InMemoryWorkflowMaterializer(_shiftyWorkflow)))
.Build();
}
/// <summary>
/// When a dependency workflow is published, all consuming workflows are updated to point to the new version of the dependency.
/// </summary>
[Fact(DisplayName = "When a workflow definition from a given source has a different Id than the one in the store, the trigger should still point to the workflow definition version ID in the store.")]
public async Task Test1()
{
// Initial population of the store from workflow providers.
await _services.PopulateRegistriesAsync();
// Artificially change the workflow definition version ID.
_shiftyWorkflow.Identity = _shiftyWorkflow.Identity with
{
Id = ":1"
};
// Emulate reloading of workflow definitions.
await _services.PopulateRegistriesAsync();
// Triggering the workflow should still work.
var stimulusSender = _services.GetRequiredService<IStimulusSender>();
var stimulus = new EventStimulus("Foo");
var triggerName = ActivityTypeNameHelper.GenerateTypeName<Event>();
var result = await stimulusSender.SendAsync(triggerName, stimulus);
Assert.NotEmpty(result.WorkflowInstanceResponses);
}
}