fix(runtime): honor transactional outbox for in-workflow PublishEvent (#8177)

* fix(runtime): route PublishEvent through transactional workflow-dispatch outbox

In-workflow async PublishEvent and IEventPublisher now dispatch via
IWorkflowDispatcher so TransactionalWorkflowDispatcher applies when
UseTransactionalOutbox is enabled. Fixes #8150.

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* test(runtime): fix PublishEvent outbox conformance tests

Register KeyValueFeature so the dispatch outbox store can persist
TriggerWorkflows items, and avoid expression-tree pattern matching in
the EventPublisher unit tests.

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

* test(runtime): assert PublishEvent outbox delivery creates consumer

ProcessAsync only queues the trigger; start the background command
processor and wait until ConsumeOrderShippedEventWorkflow is created.

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>

doc/wiki/workflow-runtime.md

@@ -69,7 +69,7 @@ Key files:
 
 ## Transactional Dispatch Outbox
 
-Hosts can opt into at-least-once workflow dispatch for dispatch calls made from inside a running workflow:
+Hosts can opt into at-least-once workflow dispatch for dispatch calls made from inside a running workflow, including child workflow dispatches and in-workflow asynchronous event publications (`PublishEvent` / `IEventPublisher.PublishAsync(..., asynchronous: true)`):
 
 ```csharp
 services.Configure<WorkflowDispatcherOptions>(options =>
@@ -86,7 +86,8 @@ Operational notes:
 - Delivery is at-least-once. If the process crashes after sending a command but before deleting the outbox record, the processor may send it again.
 - Workflow definition dispatches generated by `DispatchWorkflow`/`BulkDispatchWorkflows` include a child workflow instance ID. [DispatchWorkflowRequestHandler](../../src/modules/Elsa.Workflows.Runtime/Handlers/DispatchWorkflowRequestHandler.cs) treats that ID as the idempotency key for outbox-routed commands and skips duplicate create-and-run attempts when the instance already exists.
 - Outbox processing is serialized with the configured distributed lock provider. Poison items are abandoned after `WorkflowDispatcherOptions.MaxOutboxDeliveryAttempts`, and missing-owner items are removed after `WorkflowDispatcherOptions.OrphanedOutboxItemRetention`.
-- Dispatch calls outside a workflow execution context continue to use the regular background dispatcher.
+- Dispatch calls outside a workflow execution context, including API-triggered asynchronous events, continue to use the regular background dispatcher.
+- In-workflow `PublishEvent` and asynchronous `IEventPublisher` calls use `IWorkflowDispatcher` (`DispatchTriggerWorkflowsRequest`) so `TransactionalWorkflowDispatcher` applies. They do not go through `IStimulusDispatcher` / `BackgroundStimulusDispatcher`.
 
 ## Triggers And Bookmarks
 

src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs

@@ -16,6 +16,10 @@ public class WorkflowDispatcherOptions
     /// <summary>
     /// Gets or sets whether workflow dispatch calls made during workflow execution are written to the transactional outbox.
     /// </summary>
+    /// <remarks>
+    /// When enabled, in-workflow child dispatches and in-workflow asynchronous event publications
+    /// (<c>PublishEvent</c> / <c>IEventPublisher</c>) are written to the same outbox and delivered after the parent state commits.
+    /// </remarks>
     public bool UseTransactionalOutbox { get; set; }
 
     /// <summary>

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

@@ -1,11 +1,12 @@
 using Elsa.Workflows.Helpers;
 using Elsa.Workflows.Runtime.Activities;
+using Elsa.Workflows.Runtime.Requests;
 using Elsa.Workflows.Runtime.Stimuli;
 
 namespace Elsa.Workflows.Runtime;
 
 /// <inheritdoc />
-public class EventPublisher(IStimulusSender stimulusSender, IStimulusDispatcher stimulusDispatcher) : IEventPublisher
+public class EventPublisher(IStimulusSender stimulusSender, IWorkflowDispatcher workflowDispatcher) : IEventPublisher
 {
     /// <inheritdoc />
     public async Task PublishAsync(
@@ -22,6 +23,20 @@ public class EventPublisher(IStimulusSender stimulusSender, IStimulusDispatcher
         {
             [Event.EventInputWorkflowInputKey] = payload ?? new Dictionary<string, object>()
         };
+        var triggerName = ActivityTypeNameHelper.GenerateTypeName<Event>();
+
+        if (asynchronous)
+        {
+            await workflowDispatcher.DispatchAsync(new DispatchTriggerWorkflowsRequest(triggerName, stimulus)
+            {
+                CorrelationId = correlationId,
+                WorkflowInstanceId = workflowInstanceId,
+                ActivityInstanceId = activityInstanceId,
+                Input = workflowInput
+            }, options: null, cancellationToken);
+            return;
+        }
+
         var metadata = new StimulusMetadata
         {
             CorrelationId = correlationId,
@@ -29,17 +44,6 @@ public class EventPublisher(IStimulusSender stimulusSender, IStimulusDispatcher
             WorkflowInstanceId = workflowInstanceId,
             Input = workflowInput
         };
-        var triggerName = ActivityTypeNameHelper.GenerateTypeName<Event>();
-        if (asynchronous)
-        {
-            await stimulusDispatcher.SendAsync(new()
-            {
-                ActivityTypeName = triggerName,
-                Stimulus = stimulus,
-                Metadata = metadata
-            }, cancellationToken);
-        }
-        else
         await stimulusSender.SendAsync(triggerName, stimulus, metadata, cancellationToken);
     }
 }

test/integration/Elsa.Workflows.IntegrationTests/Scenarios/PublishEventOutbox/Tests.cs

@@ -0,0 +1,131 @@
+using Elsa.Common.Models;
+using Elsa.KeyValues.Features;
+using Elsa.Mediator.HostedServices;
+using Elsa.Testing.Shared;
+using Elsa.Workflows.Helpers;
+using Elsa.Workflows.Management;
+using Elsa.Workflows.Management.Entities;
+using Elsa.Workflows.Management.Filters;
+using Elsa.Workflows.Models;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Activities;
+using Elsa.Workflows.Runtime.Messages;
+using Elsa.Workflows.Runtime.Models;
+using Elsa.Workflows.Runtime.Options;
+using Elsa.Workflows.Runtime.Stimuli;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Xunit.Abstractions;
+
+namespace Elsa.Workflows.IntegrationTests.Scenarios.PublishEventOutbox;
+
+public class Tests
+{
+    private readonly ITestOutputHelper _testOutputHelper;
+
+    public Tests(ITestOutputHelper testOutputHelper)
+    {
+        _testOutputHelper = testOutputHelper;
+    }
+
+    [Fact(DisplayName = "With outbox on, in-workflow PublishEvent survives a crash between parent commit and delivery")]
+    public async Task PublishEvent_WithOutboxOn_SurvivesCrashBetweenCommitAndDelivery()
+    {
+        var services = CreateServices(useTransactionalOutbox: true, processOutboxAfterCommit: false);
+        await services.PopulateRegistriesAsync();
+
+        var parentResponse = await RunPublisherAsync(services);
+        Assert.Equal(WorkflowStatus.Finished, parentResponse.Status);
+
+        var outboxItems = (await services.GetRequiredService<IWorkflowDispatchOutboxStore>().FindManyAsync()).ToList();
+        var outboxItem = Assert.Single(outboxItems);
+        Assert.Equal(WorkflowDispatchOutboxItemKind.TriggerWorkflows, outboxItem.Kind);
+        Assert.Equal(parentResponse.WorkflowInstanceId, outboxItem.OwnerWorkflowInstanceId);
+        Assert.NotNull(outboxItem.TriggerWorkflowsCommand);
+        Assert.Equal(ActivityTypeNameHelper.GenerateTypeName<Event>(), outboxItem.TriggerWorkflowsCommand.ActivityTypeName);
+        Assert.Equal(PublishOrderShippedEventWorkflow.EventName, Assert.IsType<EventStimulus>(outboxItem.TriggerWorkflowsCommand.Stimulus).EventName);
+
+        var parentInstance = await services.GetRequiredService<IWorkflowInstanceStore>().FindAsync(new WorkflowInstanceFilter
+        {
+            Id = parentResponse.WorkflowInstanceId
+        });
+        Assert.NotNull(parentInstance);
+        Assert.True(parentInstance.WorkflowState.HasWorkflowDispatchOutboxItem(outboxItem.Id));
+        Assert.Empty(await FindConsumerInstancesAsync(services));
+
+        await services.GetRequiredService<IWorkflowDispatchOutboxProcessor>().ProcessAsync();
+
+        Assert.Empty(await services.GetRequiredService<IWorkflowDispatchOutboxStore>().FindManyAsync());
+
+        var commandProcessor = services.GetServices<IHostedService>().OfType<BackgroundCommandSenderHostedService>().Single();
+        await commandProcessor.StartAsync(CancellationToken.None);
+
+        try
+        {
+            await WaitUntilConsumerInstanceExistsAsync(services);
+            Assert.Single(await FindConsumerInstancesAsync(services));
+        }
+        finally
+        {
+            await commandProcessor.StopAsync(CancellationToken.None);
+        }
+    }
+
+    [Fact(DisplayName = "With outbox off, in-workflow PublishEvent is not written to the outbox")]
+    public async Task PublishEvent_WithOutboxOff_DoesNotWriteToOutbox()
+    {
+        var services = CreateServices(useTransactionalOutbox: false, processOutboxAfterCommit: false);
+        await services.PopulateRegistriesAsync();
+
+        var parentResponse = await RunPublisherAsync(services);
+        Assert.Equal(WorkflowStatus.Finished, parentResponse.Status);
+        Assert.Empty(await services.GetRequiredService<IWorkflowDispatchOutboxStore>().FindManyAsync());
+    }
+
+    private IServiceProvider CreateServices(bool useTransactionalOutbox, bool processOutboxAfterCommit)
+    {
+        return new TestApplicationBuilder(_testOutputHelper)
+            .AddWorkflow<PublishOrderShippedEventWorkflow>()
+            .AddWorkflow<ConsumeOrderShippedEventWorkflow>()
+            .ConfigureElsa(elsa => elsa.Configure<KeyValueFeature>())
+            .ConfigureServices(services => services.Configure<WorkflowDispatcherOptions>(options =>
+            {
+                options.UseTransactionalOutbox = useTransactionalOutbox;
+                options.ProcessOutboxAfterCommit = processOutboxAfterCommit;
+            }))
+            .Build();
+    }
+
+    private static async Task<RunWorkflowInstanceResponse> RunPublisherAsync(IServiceProvider services)
+    {
+        var workflowRuntime = services.GetRequiredService<IWorkflowRuntime>();
+        var workflowClient = await workflowRuntime.CreateClientAsync();
+        return await workflowClient.CreateAndRunInstanceAsync(new CreateAndRunWorkflowInstanceRequest
+        {
+            WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(nameof(PublishOrderShippedEventWorkflow), VersionOptions.Published)
+        });
+    }
+
+    private static ValueTask<IEnumerable<WorkflowInstance>> FindConsumerInstancesAsync(IServiceProvider services)
+    {
+        return services.GetRequiredService<IWorkflowInstanceStore>().FindManyAsync(new WorkflowInstanceFilter
+        {
+            DefinitionId = nameof(ConsumeOrderShippedEventWorkflow)
+        });
+    }
+
+    private static async Task WaitUntilConsumerInstanceExistsAsync(IServiceProvider services)
+    {
+        using var timeoutTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+
+        while (!timeoutTokenSource.IsCancellationRequested)
+        {
+            if ((await FindConsumerInstancesAsync(services)).Any())
+                return;
+
+            await Task.Delay(50);
+        }
+
+        Assert.Fail("ConsumeOrderShippedEventWorkflow was not created before the timeout elapsed.");
+    }
+}

test/integration/Elsa.Workflows.IntegrationTests/Scenarios/PublishEventOutbox/Workflows.cs

@@ -0,0 +1,40 @@
+using Elsa.Workflows.Activities;
+using Elsa.Workflows.Runtime.Activities;
+
+namespace Elsa.Workflows.IntegrationTests.Scenarios.PublishEventOutbox;
+
+public class PublishOrderShippedEventWorkflow : WorkflowBase
+{
+    public const string EventName = "OrderShipped";
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.Root = new Sequence
+        {
+            Activities =
+            {
+                new PublishEvent
+                {
+                    EventName = new(EventName)
+                }
+            }
+        };
+    }
+}
+
+public class ConsumeOrderShippedEventWorkflow : WorkflowBase
+{
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.Root = new Sequence
+        {
+            Activities =
+            {
+                new Event(PublishOrderShippedEventWorkflow.EventName)
+                {
+                    CanStartWorkflow = true
+                }
+            }
+        };
+    }
+}

test/unit/Elsa.Workflows.Runtime.UnitTests/Services/EventPublisherTests.cs

@@ -0,0 +1,73 @@
+using Elsa.Workflows.Helpers;
+using Elsa.Workflows.Runtime.Activities;
+using Elsa.Workflows.Runtime.Models;
+using Elsa.Workflows.Runtime.Requests;
+using Elsa.Workflows.Runtime.Stimuli;
+using NSubstitute;
+
+namespace Elsa.Workflows.Runtime.UnitTests.Services;
+
+public class EventPublisherTests
+{
+    private readonly IStimulusSender _stimulusSender = Substitute.For<IStimulusSender>();
+    private readonly IWorkflowDispatcher _workflowDispatcher = Substitute.For<IWorkflowDispatcher>();
+
+    [Fact]
+    public async Task PublishAsync_WhenAsynchronous_DispatchesThroughWorkflowDispatcher()
+    {
+        var publisher = CreatePublisher();
+        const string eventName = "OrderShipped";
+        const string correlationId = "corr-1";
+        const string workflowInstanceId = "wf-1";
+        const string activityInstanceId = "act-1";
+        var payload = new { Status = "Shipped" };
+
+        await publisher.PublishAsync(eventName, correlationId, workflowInstanceId, activityInstanceId, payload, asynchronous: true);
+
+        await _workflowDispatcher.Received(1).DispatchAsync(
+            Arg.Is<DispatchTriggerWorkflowsRequest>(request => IsMatchingTriggerRequest(request, eventName, correlationId, workflowInstanceId, activityInstanceId, payload)),
+            Arg.Any<DispatchWorkflowOptions?>(),
+            Arg.Any<CancellationToken>());
+        await _stimulusSender.DidNotReceiveWithAnyArgs().SendAsync(default!, default!, default, default);
+    }
+
+    [Fact]
+    public async Task PublishAsync_WhenSynchronous_SendsThroughStimulusSender()
+    {
+        var publisher = CreatePublisher();
+        const string eventName = "OrderShipped";
+        var payload = new Dictionary<string, object> { ["Status"] = "Shipped" };
+
+        await publisher.PublishAsync(eventName, payload: payload, asynchronous: false);
+
+        await _stimulusSender.Received(1).SendAsync(
+            ActivityTypeNameHelper.GenerateTypeName<Event>(),
+            Arg.Is<EventStimulus>(stimulus => stimulus.EventName == eventName),
+            Arg.Is<StimulusMetadata>(metadata =>
+                metadata.Input != null &&
+                metadata.Input[Event.EventInputWorkflowInputKey] == payload),
+            Arg.Any<CancellationToken>());
+        await _workflowDispatcher.DidNotReceiveWithAnyArgs().DispatchAsync(default(DispatchTriggerWorkflowsRequest)!, default, default);
+    }
+
+    private EventPublisher CreatePublisher() => new(_stimulusSender, _workflowDispatcher);
+
+    private static bool IsMatchingTriggerRequest(
+        DispatchTriggerWorkflowsRequest request,
+        string eventName,
+        string correlationId,
+        string workflowInstanceId,
+        string activityInstanceId,
+        object payload)
+    {
+        var stimulus = request.BookmarkPayload as EventStimulus;
+        return request.ActivityTypeName == ActivityTypeNameHelper.GenerateTypeName<Event>() &&
+               stimulus != null &&
+               stimulus.EventName == eventName &&
+               request.CorrelationId == correlationId &&
+               request.WorkflowInstanceId == workflowInstanceId &&
+               request.ActivityInstanceId == activityInstanceId &&
+               request.Input != null &&
+               request.Input[Event.EventInputWorkflowInputKey] == payload;
+    }
+}