Fix race condition in Distributed Workflow Runtime during same-workflow event publishing (#6538)

* Add asynchronous stimulus dispatching to workflow runtime

Introduced the BackgroundStimulusDispatcher and related components to enable asynchronous stimulus dispatching. Updated event publishing logic to support both synchronous and asynchronous delivery. These changes improve extensibility and allow better performance for handling workflow stimuli.

* Update event publishing to include isAsync parameter

Added a boolean parameter to differentiate between asynchronous and synchronous event publishing. Ensured correct value is passed based on workflow execution mode, improving clarity and functionality. Removed outdated TODO comment.

* Fix default value for Asynchronous and add activity type helper

Set the default value of the Asynchronous input property to `true` in `PublishEvent`. Additionally, utilize `ActivityTypeNameHelper` in `EventPublisher` to generate activity type names for improved consistency and clarity.

* Remove unused import from EventPublisher.cs

Eliminated the unnecessary import of Elsa.Workflows.Runtime.Requests. This cleanup helps reduce code clutter and improves maintainability.

* Set default value for Asynchronous property and update Payload.

Added a default value of `true` for the `Asynchronous` property to ensure consistency in event delivery behavior. Also, removed the unnecessary default value from the `Payload` property description for clarity.

* Simplify XML doc comment in IStimulusDispatcher interface

Refined the XML documentation for the `SendAsync` method by shortening the description and clarifying the statement. This improves readability and maintains the intent of the comment.

* Add MassTransit stimulus dispatching functionality

Introduced `DispatchStimulusRequestConsumer` and associated consumer definitions to enable stimulus dispatching. Added `MassTransitStimulusDispatcher` service and related options for configuring concurrency and endpoint behavior. Updated `MassTransitWorkflowDispatcherFeature` to register the stimulus dispatcher and allow enhanced dispatch configuration.

* Refactor stimulus dispatching to use a serialized message.

Introduced `DispatchStimulus` to encapsulate serialized requests. Updated dispatch logic to utilize `IPayloadSerializer` for serialization and deserialization, enabling message transport in a lightweight format.

* Remove Asynchronous input property from PublishEvent activity

The Asynchronous input property was removed as it was unnecessary, with a hardcoded `true` value now passed directly to the event publisher. This simplifies the code and ensures consistency in how events are published.

* Remove ProtoStringExtensions and consolidate string utilities.

Moved `EmptyIfNull` and `NullIfEmpty` methods to the existing `StringExtensions` class in `Elsa.Workflows.Core` for centralization. Updated references to use the consolidated extension methods. Deleted `ProtoStringExtensions` as it is now redundant.

* Add missing imports for Elsa.Extensions across mappers

Added `using Elsa.Extensions` to multiple mapper files in the runtime module. This ensures consistent access to shared extension methods, improving code clarity and reducing potential errors.

* Add missing Elsa.Extensions namespace to WorkflowInstance

The `using Elsa.Extensions` directive was added to ensure access to required extensions. This resolves potential issues with missing dependencies or functionality in the `WorkflowInstance` class.

src/modules/Elsa.MassTransit/ConsumerDefinitions/DispatchStimulusRequestConsumerDefinition.cs

@@ -0,0 +1,26 @@
+using Elsa.MassTransit.Consumers;
+using Elsa.MassTransit.Options;
+using JetBrains.Annotations;
+using MassTransit;
+using Microsoft.Extensions.Options;
+
+namespace Elsa.MassTransit.ConsumerDefinitions;
+
+/// <summary>
+/// Configures the endpoint for <see cref="DispatchStimulusRequestConsumer"/>
+/// </summary>
+[UsedImplicitly]
+public class DispatchStimulusRequestConsumerDefinition : ConsumerDefinition<DispatchStimulusRequestConsumer>
+{
+    /// <inheritdoc />
+    public DispatchStimulusRequestConsumerDefinition(IOptions<MassTransitWorkflowDispatcherOptions> options)
+    {
+        ConcurrentMessageLimit = options.Value.ConcurrentMessageLimit;
+    }
+
+    /// <inheritdoc />
+    protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, IConsumerConfigurator<DispatchStimulusRequestConsumer> consumerConfigurator, IRegistrationContext context)
+    {
+        endpointConfigurator.UseInMemoryOutbox(context);
+    }
+}

src/modules/Elsa.MassTransit/Consumers/DispatchStimulusRequestConsumer.cs

@@ -0,0 +1,31 @@
+using Elsa.MassTransit.Messages;
+using Elsa.Workflows;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Requests;
+using JetBrains.Annotations;
+using MassTransit;
+
+namespace Elsa.MassTransit.Consumers;
+
+[UsedImplicitly]
+public class DispatchStimulusRequestConsumer(IStimulusSender stimulusSender, IPayloadSerializer payloadSerializer) : IConsumer<DispatchStimulus>
+{
+    public async Task Consume(ConsumeContext<DispatchStimulus> context)
+    {
+        var cancellationToken = context.CancellationToken;
+        var message = context.Message;
+        var json = message.SerializedRequest;
+        var request = payloadSerializer.Deserialize<DispatchStimulusRequest>(json);
+        
+        if(request.ActivityTypeName != null)
+        {
+            await stimulusSender.SendAsync(request.ActivityTypeName, request.Stimulus!, request.Metadata, cancellationToken);
+            return;
+        }
+        
+        if(request.StimulusHash != null)
+            await stimulusSender.SendAsync(request.StimulusHash!, request.Metadata, cancellationToken);
+        
+        throw new InvalidOperationException("Either ActivityTypeName or StimulusHash must be specified.");
+    }
+}

src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs

@@ -23,18 +23,33 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase
     public MassTransitWorkflowDispatcherFeature(IModule module) : base(module)
     {
     }
+
+    /// <summary>
+    /// Configures the MassTransit workflow dispatcher.
+    /// </summary>
+    [Obsolete("Use ConfigureWorkflowDispatcherOptions instead.")]
+    public Action<MassTransitWorkflowDispatcherOptions> ConfigureDispatcherOptions
+    {
+        get => ConfigureWorkflowDispatcherOptions; 
+        set => ConfigureWorkflowDispatcherOptions = value;
+    }
     
     /// <summary>
     /// Configures the MassTransit workflow dispatcher.
     /// </summary>
-    public Action<MassTransitWorkflowDispatcherOptions>? ConfigureDispatcherOptions { get; set; }
+    public Action<MassTransitWorkflowDispatcherOptions> ConfigureWorkflowDispatcherOptions { get; set; } = _ => { };
     
+    /// <summary>
+    /// Configures the MassTransit stimulus dispatcher.
+    /// </summary>
+    public Action<MassTransitStimulusDispatcherOptions> ConfigureStimulusDispatcherOptions { get; set; } = _ => { };
 
     /// <inheritdoc />
     public override void Configure()
     {
         Module.AddMassTransitConsumer<DispatchWorkflowRequestConsumer, DispatchWorkflowRequestConsumerDefinition>();
         Module.AddMassTransitConsumer<DispatchCancelWorkflowsRequestConsumer>("elsa-dispatch-cancel-workflow", true);
+        Module.AddMassTransitConsumer<DispatchStimulusRequestConsumer, DispatchStimulusRequestConsumerDefinition>("elsa-dispatch-stimulus");
         Module.Configure<WorkflowRuntimeFeature>(f =>
         {
             f.WorkflowDispatcher = sp =>
@@ -44,17 +59,17 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase
             };
 
             f.WorkflowCancellationDispatcher = sp => sp.GetRequiredService<MassTransitWorkflowCancellationDispatcher>();
+            f.StimulusDispatcher = sp => sp.GetRequiredService<MassTransitStimulusDispatcher>();
         });
     }
 
     /// <inheritdoc />
     public override void Apply()
     {
-        var options = Services.AddOptions<MassTransitWorkflowDispatcherOptions>();
-        
-        if (ConfigureDispatcherOptions != null)
-            options.Configure(ConfigureDispatcherOptions);
+        Services.Configure(ConfigureWorkflowDispatcherOptions);
+        Services.Configure(ConfigureStimulusDispatcherOptions);
         
         Services.AddScoped<MassTransitWorkflowCancellationDispatcher>();
+        Services.AddScoped<MassTransitStimulusDispatcher>();
     }
 }

src/modules/Elsa.MassTransit/Messages/DispatchStimulus.cs

@@ -0,0 +1,3 @@
+namespace Elsa.MassTransit.Messages;
+
+public record DispatchStimulus(string SerializedRequest);

src/modules/Elsa.MassTransit/Options/MassTransitStimulusDispatcherOptions.cs

@@ -0,0 +1,14 @@
+using Elsa.MassTransit.ConsumerDefinitions;
+
+namespace Elsa.MassTransit.Options;
+
+/// <summary>
+/// Provides options to the <see cref="DispatchStimulusRequestConsumerDefinition"/>
+/// </summary>
+public class MassTransitStimulusDispatcherOptions
+{
+    /// <summary>
+    /// The number of concurrent messages to process.
+    /// </summary>
+    public int? ConcurrentMessageLimit { get; set; }
+}

src/modules/Elsa.MassTransit/Services/MassTransitStimulusDispatcher.cs

@@ -0,0 +1,19 @@
+using Elsa.MassTransit.Messages;
+using Elsa.Workflows;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Requests;
+using Elsa.Workflows.Runtime.Responses;
+using MassTransit;
+
+namespace Elsa.MassTransit.Services;
+
+public class MassTransitStimulusDispatcher(IBus bus, IPayloadSerializer payloadSerializer) : IStimulusDispatcher
+{
+    public async Task<DispatchStimulusResponse> SendAsync(DispatchStimulusRequest request, CancellationToken cancellationToken = default)
+    {
+        var json = payloadSerializer.Serialize(request);
+        var message = new DispatchStimulus(json);
+        await bus.Publish(message, cancellationToken);
+        return DispatchStimulusResponse.Empty;
+    }
+}

src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs

@@ -25,7 +25,6 @@ public class MassTransitWorkflowDispatcher(
     IStimulusHasher stimulusHasher,
     ITriggerStore triggerStore,
     IBookmarkStore bookmarkStore,
-    IPayloadSerializer jsonSerializer,
     ILogger<MassTransitWorkflowDispatcher> logger)
     : IWorkflowDispatcher
 {
@@ -174,9 +173,4 @@ public class MassTransitWorkflowDispatcher(
         var sendEndpoint = await bus.GetSendEndpoint(new($"queue:{endpointName}"));
         return sendEndpoint;
     }
-    
-    private string? SerializeInput(object? input)
-    {
-        return input != null ? jsonSerializer.Serialize(input) : null;
-    }
 }

src/modules/Elsa.Workflows.Api/Endpoints/Events/TriggerAuthenticated/Endpoint.cs

@@ -38,15 +38,14 @@ internal class Trigger : ElsaEndpoint<Request>
 
         if (workflowExecutionMode == WorkflowExecutionMode.Asynchronous)
         {
-            // TODO: Implement async dispatch
-            await _eventPublisher.PublishAsync(eventName, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken);
+            await _eventPublisher.PublishAsync(eventName, correlationId, workflowInstanceId, activityInstanceId, input, true, cancellationToken);
             await SendOkAsync(cancellationToken);
         }
         else
         {
-            await _eventPublisher.PublishAsync(eventName, correlationId, workflowInstanceId, activityInstanceId, input, cancellationToken);
+            await _eventPublisher.PublishAsync(eventName, correlationId, workflowInstanceId, activityInstanceId, input, false, cancellationToken);
 
-            if (!HttpContext.Response.HasStarted) 
+            if (!HttpContext.Response.HasStarted)
                 await SendOkAsync(cancellationToken);
         }
     }

src/modules/Elsa.Workflows.Core/Extensions/StringExtensions.cs

@@ -4,4 +4,6 @@ namespace Elsa.Extensions;
 public static class StringExtensions
 {
     public static string WithDefault(this string? value, string defaultValue) => !string.IsNullOrWhiteSpace(value) ? value : defaultValue;
+    public static string EmptyIfNull(this string? value) => value ?? "";
+    public static string? NullIfEmpty(this string? value) => value == "" ? null : value;
 }

src/modules/Elsa.Workflows.Runtime.ProtoActor/Actors/WorkflowInstance.cs

@@ -1,3 +1,4 @@
+using Elsa.Extensions;
 using Elsa.Workflows.Management;
 using Elsa.Workflows.Management.Options;
 using Elsa.Workflows.Models;

src/modules/Elsa.Workflows.Runtime.ProtoActor/Extensions/ProtoStringExtensions.cs

@@ -1,7 +0,0 @@
-namespace Elsa.Workflows.Runtime.ProtoActor.Extensions;
-
-internal static class ProtoStringExtensions
-{
-    public static string EmptyIfNull(this string? value) => value ?? "";
-    public static string? NullIfEmpty(this string? value) => value == "" ? default : value;
-}

src/modules/Elsa.Workflows.Runtime.ProtoActor/Mappers/ActivityHandleMapper.cs

@@ -1,3 +1,4 @@
+using Elsa.Extensions;
 using Elsa.Workflows.Runtime.ProtoActor.Extensions;
 using Elsa.Workflows.Models;
 using ProtoActivityHandle = Elsa.Workflows.Runtime.ProtoActor.ProtoBuf.ActivityHandle;

src/modules/Elsa.Workflows.Runtime.ProtoActor/Mappers/CreateAndRunWorkflowInstanceRequestMapper.cs

@@ -1,3 +1,4 @@
+using Elsa.Extensions;
 using Elsa.Workflows.Runtime.ProtoActor.Extensions;
 using Elsa.Workflows.Runtime.Messages;
 using ProtoCreateAndRunWorkflowInstanceRequest = Elsa.Workflows.Runtime.ProtoActor.ProtoBuf.CreateAndRunWorkflowInstanceRequest;

src/modules/Elsa.Workflows.Runtime.ProtoActor/Mappers/CreateWorkflowInstanceRequestMapper.cs

@@ -1,3 +1,4 @@
+using Elsa.Extensions;
 using Elsa.Workflows.Runtime.ProtoActor.Extensions;
 using Elsa.Workflows.Runtime.Messages;
 using ProtoCreateWorkflowInstanceRequest = Elsa.Workflows.Runtime.ProtoActor.ProtoBuf.CreateWorkflowInstanceRequest;

src/modules/Elsa.Workflows.Runtime.ProtoActor/Mappers/RunWorkflowInstanceRequestMapper.cs

@@ -1,3 +1,4 @@
+using Elsa.Extensions;
 using Elsa.Workflows.Runtime.ProtoActor.Extensions;
 using Elsa.Workflows.Runtime.Messages;
 using ProtoRunWorkflowInstanceRequest = Elsa.Workflows.Runtime.ProtoActor.ProtoBuf.RunWorkflowInstanceRequest;

src/modules/Elsa.Workflows.Runtime.ProtoActor/Mappers/RunWorkflowParamsMapper.cs

@@ -1,3 +1,4 @@
+using Elsa.Extensions;
 using Elsa.Workflows.Runtime.ProtoActor.Extensions;
 using Elsa.Workflows.Options;
 using ProtoRunWorkflowInstanceRequest = Elsa.Workflows.Runtime.ProtoActor.ProtoBuf.RunWorkflowInstanceRequest;

src/modules/Elsa.Workflows.Runtime.ProtoActor/Mappers/WorkflowDefinitionHandleMapper.cs

@@ -1,4 +1,5 @@
 using Elsa.Common.Models;
+using Elsa.Extensions;
 using Elsa.Workflows.Runtime.ProtoActor.Extensions;
 using Elsa.Workflows.Models;
 using ProtoWorkflowDefinitionHandle = Elsa.Workflows.Runtime.ProtoActor.ProtoBuf.WorkflowDefinitionHandle;

src/modules/Elsa.Workflows.Runtime/Activities/PublishEvent.cs

@@ -11,13 +11,8 @@ namespace Elsa.Workflows.Runtime.Activities;
 /// </summary>
 [Activity("Elsa", "Primitives", "Publishes an event.")]
 [UsedImplicitly]
-public class PublishEvent : Activity
+public class PublishEvent([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : Activity(source, line)
 {
-    /// <inheritdoc />
-    public PublishEvent([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
-    {
-    }
-
     /// <summary>
     /// The name of the event to publish.
     /// </summary>
@@ -46,13 +41,13 @@ public class PublishEvent : Activity
     protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
     {
         var eventName = EventName.Get(context);
-        var correlationId = CorrelationId.GetOrDefault(context);
+        var correlationId = CorrelationId.GetOrDefault(context).NullIfEmpty();
         var isLocalEvent = IsLocalEvent.GetOrDefault(context);
         var workflowInstanceId = isLocalEvent ? context.WorkflowExecutionContext.Id : null;
         var payload = Payload.GetOrDefault(context);
         var publisher = context.GetRequiredService<IEventPublisher>();
-        
-        await publisher.PublishAsync(eventName, correlationId, workflowInstanceId, null, payload, context.CancellationToken);
+
+        await publisher.PublishAsync(eventName, correlationId, workflowInstanceId, null, payload, true, context.CancellationToken);
         await context.CompleteActivityAsync();
     }
 }

src/modules/Elsa.Workflows.Runtime/Commands/DispatchStimulusCommand.cs

@@ -0,0 +1,7 @@
+using Elsa.Mediator.Contracts;
+using Elsa.Mediator.Models;
+using Elsa.Workflows.Runtime.Requests;
+
+namespace Elsa.Workflows.Runtime.Commands;
+
+public record DispatchStimulusCommand(DispatchStimulusRequest Request) : ICommand<Unit>;

src/modules/Elsa.Workflows.Runtime/Commands/DispatchWorkflowDefinitionCommand.cs

@@ -15,4 +15,4 @@ public class DispatchWorkflowDefinitionCommand(string definitionVersionId) : ICo
     public string? CorrelationId { get; set; }
     public string? InstanceId { get; set; }
     public string? TriggerActivityId { get; set; }
-}
+}

src/modules/Elsa.Workflows.Runtime/Contracts/IEventPublisher.cs

@@ -10,5 +10,5 @@ public interface IEventPublisher
     /// <summary>
     /// Publishes the specified event.
     /// </summary>
-    Task PublishAsync(string eventName, string? correlationId = default, string? workflowInstanceId = default, string? activityInstanceId = default, object? payload = default, CancellationToken cancellationToken = default);
+    Task PublishAsync(string eventName, string? correlationId = null, string? workflowInstanceId = null, string? activityInstanceId = null, object? payload = null, bool asynchronous = false, CancellationToken cancellationToken = default);
 }

src/modules/Elsa.Workflows.Runtime/Contracts/IStimulusDispatcher.cs

@@ -0,0 +1,12 @@
+using Elsa.Workflows.Runtime.Requests;
+using Elsa.Workflows.Runtime.Responses;
+
+namespace Elsa.Workflows.Runtime;
+
+public interface IStimulusDispatcher
+{
+    /// <summary>
+    /// Dispatches stimulus. This could result in new workflow instances being started as well as existing workflow instances being resumed.
+    /// </summary>
+    Task<DispatchStimulusResponse> SendAsync(DispatchStimulusRequest request, CancellationToken cancellationToken = default);
+}

src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs

@@ -53,6 +53,11 @@ public class WorkflowRuntimeFeature(IModule module) : FeatureBase(module)
         var decoratedService = ActivatorUtilities.CreateInstance<BackgroundWorkflowDispatcher>(sp);
         return ActivatorUtilities.CreateInstance<ValidatingWorkflowDispatcher>(sp, decoratedService);
     };
+    
+    /// <summary>
+    /// A factory that instantiates an <see cref="IStimulusDispatcher"/>.
+    /// </summary>
+    public Func<IServiceProvider, IStimulusDispatcher> StimulusDispatcher { get; set; } = sp => ActivatorUtilities.CreateInstance<BackgroundStimulusDispatcher>(sp);
 
     /// <summary>
     /// A factory that instantiates an <see cref="IWorkflowCancellationDispatcher"/>.
@@ -237,6 +242,7 @@ public class WorkflowRuntimeFeature(IModule module) : FeatureBase(module)
             .AddScoped<IBackgroundActivityInvoker, BackgroundActivityInvoker>()
             .AddScoped(WorkflowRuntime)
             .AddScoped(WorkflowDispatcher)
+            .AddScoped(StimulusDispatcher)
             .AddScoped(WorkflowCancellationDispatcher)
             .AddScoped(RunTaskDispatcher)
             .AddScoped(ActivityExecutionLogSink)
@@ -324,6 +330,7 @@ public class WorkflowRuntimeFeature(IModule module) : FeatureBase(module)
 
             // Domain handlers.
             .AddCommandHandler<DispatchWorkflowCommandHandler>()
+            .AddCommandHandler<DispatchStimulusCommandHandler>()
             .AddCommandHandler<CancelWorkflowsCommandHandler>()
             .AddNotificationHandler<ResumeDispatchWorkflowActivity>()
             .AddNotificationHandler<ResumeBulkDispatchWorkflowActivity>()

src/modules/Elsa.Workflows.Runtime/Handlers/DispatchStimulusCommandHandler.cs

@@ -0,0 +1,26 @@
+using Elsa.Mediator.Contracts;
+using Elsa.Mediator.Models;
+using Elsa.Workflows.Runtime.Commands;
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.Runtime.Handlers;
+
+[UsedImplicitly]
+public class DispatchStimulusCommandHandler(IStimulusSender stimulusSender) : ICommandHandler<DispatchStimulusCommand>
+{
+    public async Task<Unit> HandleAsync(DispatchStimulusCommand command, CancellationToken cancellationToken)
+    {
+        var request = command.Request;
+        
+        if(request.ActivityTypeName != null)
+        {
+            await stimulusSender.SendAsync(request.ActivityTypeName, request.Stimulus!, request.Metadata, cancellationToken);
+            return Unit.Instance;
+        }
+        
+        if(request.StimulusHash != null)
+            await stimulusSender.SendAsync(request.StimulusHash!, request.Metadata, cancellationToken);
+        
+        throw new InvalidOperationException("Either ActivityTypeName or StimulusHash must be specified.");
+    }
+}

src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsRequest.cs

@@ -3,7 +3,7 @@ namespace Elsa.Workflows.Runtime.Requests;
 public record DispatchResumeWorkflowsRequest(
     string ActivityTypeName, 
     object BookmarkPayload, 
-    string? CorrelationId = default,
-    string? WorkflowInstanceId = default,
-    string? ActivityInstanceId = default,
-    IDictionary<string, object>? Input = default);
+    string? CorrelationId = null,
+    string? WorkflowInstanceId = null,
+    string? ActivityInstanceId = null,
+    IDictionary<string, object>? Input = null);

src/modules/Elsa.Workflows.Runtime/Requests/DispatchStimulusRequest.cs

@@ -0,0 +1,9 @@
+namespace Elsa.Workflows.Runtime.Requests;
+
+public class DispatchStimulusRequest
+{
+    public string? ActivityTypeName { get; set; }
+    public object? Stimulus { get; set; }
+    public string? StimulusHash { get; set; }
+    public StimulusMetadata? Metadata { get; set; }
+}

src/modules/Elsa.Workflows.Runtime/Responses/DispatchStimulusResponse.cs

@@ -0,0 +1,6 @@
+namespace Elsa.Workflows.Runtime.Responses;
+
+public record DispatchStimulusResponse
+{
+    public static DispatchStimulusResponse Empty { get; } = new();
+}

src/modules/Elsa.Workflows.Runtime/Services/BackgroundStimulusDispatcher.cs

@@ -0,0 +1,21 @@
+using Elsa.Mediator;
+using Elsa.Mediator.Contracts;
+using Elsa.Workflows.Runtime.Commands;
+using Elsa.Workflows.Runtime.Requests;
+using Elsa.Workflows.Runtime.Responses;
+
+namespace Elsa.Workflows.Runtime;
+
+/// <summary>
+/// A simple implementation that queues the specified request for delivering stimuli on a non-durable background worker.
+/// </summary>
+public class BackgroundStimulusDispatcher(ICommandSender commandSender) : IStimulusDispatcher
+{
+    /// <inheritdoc />
+    public async Task<DispatchStimulusResponse> SendAsync(DispatchStimulusRequest request, CancellationToken cancellationToken = default)
+    {
+        var command = new DispatchStimulusCommand(request);
+        await commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken);
+        return DispatchStimulusResponse.Empty;
+    }
+}

src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs

@@ -1,10 +1,11 @@
+using Elsa.Workflows.Helpers;
 using Elsa.Workflows.Runtime.Activities;
 using Elsa.Workflows.Runtime.Stimuli;
 
 namespace Elsa.Workflows.Runtime;
 
 /// <inheritdoc />
-public class EventPublisher(IStimulusSender stimulusSender) : IEventPublisher
+public class EventPublisher(IStimulusSender stimulusSender, IStimulusDispatcher stimulusDispatcher) : IEventPublisher
 {
     /// <inheritdoc />
     public async Task PublishAsync(
@@ -13,6 +14,7 @@ public class EventPublisher(IStimulusSender stimulusSender) : IEventPublisher
         string? workflowInstanceId = null,
         string? activityInstanceId = null,
         object? payload = null,
+        bool asynchronous = false,
         CancellationToken cancellationToken = default)
     {
         var stimulus = new EventStimulus(eventName);
@@ -27,6 +29,16 @@ public class EventPublisher(IStimulusSender stimulusSender) : IEventPublisher
             WorkflowInstanceId = workflowInstanceId,
             Input = workflowInput
         };
-        await stimulusSender.SendAsync<Event>(stimulus, metadata, cancellationToken);
+        if (asynchronous)
+        {
+            await stimulusDispatcher.SendAsync(new()
+            {
+                ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<Event>(),
+                Stimulus = stimulus,
+                Metadata = metadata
+            }, cancellationToken);
+        }
+        else
+            await stimulusSender.SendAsync<Event>(stimulus, metadata, cancellationToken);
     }
 }