diff --git a/doc/wiki/workflow-runtime.md b/doc/wiki/workflow-runtime.md index f75f96bf3..7f27b15ff 100644 --- a/doc/wiki/workflow-runtime.md +++ b/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(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 diff --git a/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs index c804c8f84..f7160a5b2 100644 --- a/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs @@ -16,6 +16,10 @@ public class WorkflowDispatcherOptions /// /// Gets or sets whether workflow dispatch calls made during workflow execution are written to the transactional outbox. /// + /// + /// When enabled, in-workflow child dispatches and in-workflow asynchronous event publications + /// (PublishEvent / IEventPublisher) are written to the same outbox and delivered after the parent state commits. + /// public bool UseTransactionalOutbox { get; set; } /// diff --git a/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs b/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs index a760accbe..88d1bafb9 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/EventPublisher.cs +++ b/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; /// -public class EventPublisher(IStimulusSender stimulusSender, IStimulusDispatcher stimulusDispatcher) : IEventPublisher +public class EventPublisher(IStimulusSender stimulusSender, IWorkflowDispatcher workflowDispatcher) : IEventPublisher { /// public async Task PublishAsync( @@ -22,6 +23,20 @@ public class EventPublisher(IStimulusSender stimulusSender, IStimulusDispatcher { [Event.EventInputWorkflowInputKey] = payload ?? new Dictionary() }; + var triggerName = ActivityTypeNameHelper.GenerateTypeName(); + + 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(); - if (asynchronous) - { - await stimulusDispatcher.SendAsync(new() - { - ActivityTypeName = triggerName, - Stimulus = stimulus, - Metadata = metadata - }, cancellationToken); - } - else - await stimulusSender.SendAsync(triggerName, stimulus, metadata, cancellationToken); + await stimulusSender.SendAsync(triggerName, stimulus, metadata, cancellationToken); } } \ No newline at end of file diff --git a/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/PublishEventOutbox/Tests.cs b/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/PublishEventOutbox/Tests.cs new file mode 100644 index 000000000..414bf9e51 --- /dev/null +++ b/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().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(), outboxItem.TriggerWorkflowsCommand.ActivityTypeName); + Assert.Equal(PublishOrderShippedEventWorkflow.EventName, Assert.IsType(outboxItem.TriggerWorkflowsCommand.Stimulus).EventName); + + var parentInstance = await services.GetRequiredService().FindAsync(new WorkflowInstanceFilter + { + Id = parentResponse.WorkflowInstanceId + }); + Assert.NotNull(parentInstance); + Assert.True(parentInstance.WorkflowState.HasWorkflowDispatchOutboxItem(outboxItem.Id)); + Assert.Empty(await FindConsumerInstancesAsync(services)); + + await services.GetRequiredService().ProcessAsync(); + + Assert.Empty(await services.GetRequiredService().FindManyAsync()); + + var commandProcessor = services.GetServices().OfType().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().FindManyAsync()); + } + + private IServiceProvider CreateServices(bool useTransactionalOutbox, bool processOutboxAfterCommit) + { + return new TestApplicationBuilder(_testOutputHelper) + .AddWorkflow() + .AddWorkflow() + .ConfigureElsa(elsa => elsa.Configure()) + .ConfigureServices(services => services.Configure(options => + { + options.UseTransactionalOutbox = useTransactionalOutbox; + options.ProcessOutboxAfterCommit = processOutboxAfterCommit; + })) + .Build(); + } + + private static async Task RunPublisherAsync(IServiceProvider services) + { + var workflowRuntime = services.GetRequiredService(); + var workflowClient = await workflowRuntime.CreateClientAsync(); + return await workflowClient.CreateAndRunInstanceAsync(new CreateAndRunWorkflowInstanceRequest + { + WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(nameof(PublishOrderShippedEventWorkflow), VersionOptions.Published) + }); + } + + private static ValueTask> FindConsumerInstancesAsync(IServiceProvider services) + { + return services.GetRequiredService().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."); + } +} diff --git a/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/PublishEventOutbox/Workflows.cs b/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/PublishEventOutbox/Workflows.cs new file mode 100644 index 000000000..41a85cb17 --- /dev/null +++ b/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 + } + } + }; + } +} diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/EventPublisherTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/EventPublisherTests.cs new file mode 100644 index 000000000..fecc68e6c --- /dev/null +++ b/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(); + private readonly IWorkflowDispatcher _workflowDispatcher = Substitute.For(); + + [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(request => IsMatchingTriggerRequest(request, eventName, correlationId, workflowInstanceId, activityInstanceId, payload)), + Arg.Any(), + Arg.Any()); + 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 { ["Status"] = "Shipped" }; + + await publisher.PublishAsync(eventName, payload: payload, asynchronous: false); + + await _stimulusSender.Received(1).SendAsync( + ActivityTypeNameHelper.GenerateTypeName(), + Arg.Is(stimulus => stimulus.EventName == eventName), + Arg.Is(metadata => + metadata.Input != null && + metadata.Input[Event.EventInputWorkflowInputKey] == payload), + Arg.Any()); + 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() && + stimulus != null && + stimulus.EventName == eventName && + request.CorrelationId == correlationId && + request.WorkflowInstanceId == workflowInstanceId && + request.ActivityInstanceId == activityInstanceId && + request.Input != null && + request.Input[Event.EventInputWorkflowInputKey] == payload; + } +}