Updated output handling of asynchronously run workflows to be the same as when run synchronously

src/common/Elsa.Testing.Shared.Integration/TestWorkflow.cs

@@ -0,0 +1,18 @@
+using Elsa.Workflows;
+
+namespace Elsa.Testing.Shared;
+
+public class TestWorkflow : WorkflowBase
+{
+    private readonly Action<IWorkflowBuilder> _buildWorkflow;
+
+    public TestWorkflow(Action<IWorkflowBuilder> buildWorkflow)
+    {
+        _buildWorkflow = buildWorkflow;
+    }
+
+    protected override void Build(IWorkflowBuilder workflowBuilder)
+    {
+        _buildWorkflow(workflowBuilder);
+    }
+}

src/common/Elsa.Testing.Shared.Integration/WorkflowExtensions.cs

@@ -0,0 +1,89 @@
+using Elsa.Extensions;
+using Elsa.Features.Services;
+using Elsa.Mediator.Contracts;
+using Elsa.Workflows;
+using Elsa.Workflows.Notifications;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Contracts;
+using Elsa.Workflows.Runtime.Requests;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+
+namespace Elsa.Testing.Shared;
+
+public static class WorkflowExtensions
+{
+    public static async Task<WorkflowFinished?> DispatchWorkflowAndRunToCompletion(
+        this IWorkflow workflowDefinition,
+        Action<IServiceCollection>? configureServices = null,
+        Action<IModule>? configureElsa = null,
+        string? instanceid = null,
+        string? definitionVersionId = null,
+        TimeSpan? timeout = default)
+    {
+        SemaphoreSlim semaphore = new (0, 1);
+        WorkflowFinished? workflowFinishedRecord = null;
+
+        var host = Host.CreateDefaultBuilder()
+            .ConfigureServices(services =>
+            {
+                configureServices?.Invoke(services);
+
+                // This notification handler will capture the WorkflowFinished record (to be returned) and release the semaphore
+                services.AddNotificationHandler<WorkflowFinishedAction, WorkflowFinished>(sp => new WorkflowFinishedAction(notification =>
+                {
+                    workflowFinishedRecord = notification;
+                    semaphore.Release();
+                }
+                ));
+
+                services.AddElsa(elsa => {
+                    configureElsa?.Invoke(elsa);
+                });
+            })
+            .Build();
+
+        try
+        {
+            // Start the Host
+            await host.StartAsync(CancellationToken.None);
+
+            await host.Services.PopulateRegistriesAsync();
+
+            // Build the workflow
+            IWorkflowBuilderFactory workflowBuilderFactory = host.Services.GetRequiredService<IWorkflowBuilderFactory>();
+            var workflow = await workflowBuilderFactory.CreateBuilder().BuildWorkflowAsync(workflowDefinition);
+
+            // Register the workflow
+            var workflowRegistry = host.Services.GetRequiredService<IWorkflowRegistry>();
+            await workflowRegistry.RegisterAsync(workflow);
+
+            // Dispatch the workflow
+            var workflowDispatcher = host.Services.GetRequiredService<IWorkflowDispatcher>();
+            var dispatchWorkflowResponse = await workflowDispatcher.DispatchAsync(new DispatchWorkflowDefinitionRequest()
+            {
+                DefinitionVersionId = "sampleWorkflow",
+                InstanceId = "instanceId"
+            });
+            dispatchWorkflowResponse.ThrowIfFailed();
+
+            // Wait for the workflow to complete, and then return the WorkflowFinished notification
+            bool signaled = await semaphore.WaitAsync(timeout ?? TimeSpan.FromSeconds(5));
+            return signaled ? workflowFinishedRecord : null;
+        }
+        finally
+        {
+            // Stop the Host
+            await host.StopAsync(CancellationToken.None);
+        }
+    }
+
+    class WorkflowFinishedAction(Action<WorkflowFinished> action) : INotificationHandler<WorkflowFinished>
+    {
+        public Task HandleAsync(WorkflowFinished notification, CancellationToken cancellationToken)
+        {
+            action(notification);
+            return Task.CompletedTask;
+        }
+    }
+}

src/modules/Elsa.Workflows.Runtime/Middleware/Activities/BackgroundActivityInvokerMiddleware.cs

@@ -122,7 +122,7 @@ private static void CaptureOutputIfAny(ActivityExecutionContext context)
                 continue;
 
             var output = (Output?)outputDescriptor.ValueGetter(activity);
-            context.Set(output, outputEntry.Value);
+            context.Set(output, outputEntry.Value, outputDescriptor.Name);
         }
     }
 

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

@@ -90,35 +90,8 @@ private async Task ResumeWorkflowAsync(ActivityExecutionContext activityExecutio
     }
 
     private IDictionary<string, object> ExtractActivityOutput(ActivityExecutionContext activityExecutionContext)
-    {
-        var outputDescriptors = activityExecutionContext.ActivityDescriptor.Outputs;
-        var outputValues = new Dictionary<string, object>();
-
-        foreach (var outputDescriptor in outputDescriptors)
-        {
-            var output = (Output?)outputDescriptor.ValueGetter(activityExecutionContext.Activity);
-
-            if (output == null)
-                continue;
-
-            var memoryBlockReference = output.MemoryBlockReference();
-
-            if (!activityExecutionContext.ExpressionExecutionContext.TryGetBlock(memoryBlockReference, out var memoryBlock))
-                continue;
-
-            var variableMetadata = memoryBlock.Metadata as VariableBlockMetadata;
-            var driver = variableMetadata?.StorageDriverType;
-
-            // We only capture output written to the workflow itself. Other drivers like blob storage, etc. will be ignored since the foreground context will be loading those.
-            if (driver != typeof(WorkflowStorageDriver) && driver != typeof(WorkflowInstanceStorageDriver) && driver != null)
-                continue;
-
-            var outputValue = activityExecutionContext.Get(memoryBlockReference);
-
-            if (outputValue != null)
-                outputValues[outputDescriptor.Name] = outputValue;
-        }
-
-        return outputValues;
+    {
+        var activityOutputRegister = activityExecutionContext.WorkflowExecutionContext.GetActivityOutputRegister();
+        return activityOutputRegister.FindMany(a => a.ActivityInstanceId == activityExecutionContext.Id).ToDictionary(o => o.OutputName, o => o.Value!);
     }
 }

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

@@ -1,5 +1,4 @@
 using Elsa.Testing.Shared;
-using Elsa.Workflows.IntegrationTests.Scenarios.JsonObjectToObjectRemainsJsonObject.Workflows;
 using Microsoft.Extensions.DependencyInjection;
 using Xunit.Abstractions;
 
@@ -22,7 +21,7 @@ public JsonObjectJintTests(ITestOutputHelper testOutputHelper)
     public async Task Test1()
     {
         await _services.PopulateRegistriesAsync();
-        await _workflowRunner.RunAsync<TestWorkflow>();
+        await _workflowRunner.RunAsync<Workflows.TestWorkflow>();
         var lines = _capturingTextWriter.Lines.ToList();
         Assert.Equal(new[] { "Baz" }, lines);
     }

test/integration/Elsa.Workflows.IntegrationTests/Scenarios/RunAsynchronousActivityOutput/Activities/SampleActivity.cs

@@ -0,0 +1,28 @@
+using Elsa.Workflows.Attributes;
+using Elsa.Workflows.Models;
+
+namespace Elsa.Workflows.IntegrationTests.Scenarios.RunAsynchronousActivityOutput.Activities;
+
+[Activity("Elsa.Test", Kind = ActivityKind.Task)]
+internal class SampleActivity : CodeActivity
+{
+
+    public SampleActivity()
+    {
+        RunAsynchronously = true;
+    }
+
+    public Input<int>? Number1 { get; set; } = default;
+    public Input<int>? Number2 { get; set; } = default;
+    public Output<int>? Sum { get; set; } = default;
+    public Output<int>? Product { get; set; } = default;
+
+    protected override void Execute(ActivityExecutionContext context)
+    {
+        int number1 = context.Get(Number1);
+        int number2 = context.Get(Number2);
+
+        context.Set(Sum, number1 + number2);
+        context.Set(Product, number1 * number2);
+    }
+}

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

@@ -0,0 +1,164 @@
+using Elsa.Common.Services;
+using Elsa.Extensions;
+using Elsa.Testing.Shared;
+using Elsa.Workflows.Activities;
+using Elsa.Workflows.IntegrationTests.Scenarios.RunAsynchronousActivityOutput.Activities;
+using Elsa.Workflows.Memory;
+using Elsa.Workflows.Notifications;
+using Elsa.Workflows.Runtime.Distributed;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Stores;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.Workflows.IntegrationTests.Scenarios.RunAsynchronousActivityOutput;
+
+public class Tests
+{
+    [Theory(DisplayName = "Activity outputs captured in activity execution record")]
+    [InlineData(true)]
+    [InlineData(false)]
+    public async Task ActivityOutputCaptureTest(bool runAsynchronously)
+    {
+        // Arrange
+
+        TestWorkflow workflow = new(workflowBuilder =>
+            {
+                workflowBuilder.DefinitionId = "sampleWorkflow";
+                workflowBuilder.Id = "sampleWorkflow";
+
+                var variable1 = new Variable<int>();
+                workflowBuilder.Root = new Sequence
+                {
+                    Variables =
+                    {
+                        variable1
+                    },
+                    Activities =
+                    {
+                        new SampleActivity()
+                        {
+                            Id = "SampleActivity1",
+                            RunAsynchronously = runAsynchronously,
+                            Number1 = new(4),
+                            Number2 = new(8),
+                            Sum = new(variable1),
+                            Product = null
+                        }
+                    }
+                };
+            }
+        );
+
+        var activityExecutionStore = new MemoryActivityExecutionStore(new MemoryStore<ActivityExecutionRecord>());
+
+        // Act
+
+        WorkflowFinished? workflowFinishedRecord = await workflow.DispatchWorkflowAndRunToCompletion(
+            configureElsa: elsa =>
+            {
+                elsa.UseWorkflowRuntime(workflowRuntime => {
+                    workflowRuntime.ActivityExecutionLogStore = sp => activityExecutionStore;
+                });
+            }
+        );
+
+        // Assert
+
+        Assert.NotNull(workflowFinishedRecord);
+        Assert.Equal(WorkflowStatus.Finished, workflowFinishedRecord.WorkflowState.Status);
+        Assert.Equal(WorkflowSubStatus.Finished, workflowFinishedRecord.WorkflowState.SubStatus);
+
+        var activityExecutionRecord = await activityExecutionStore.FindAsync(new() { ActivityId = "SampleActivity1" });
+        Assert.NotNull(activityExecutionRecord?.Outputs);
+        Assert.Equal(2, activityExecutionRecord.Outputs!.Count);
+        Assert.Equal(12, activityExecutionRecord.Outputs!.GetValue<int>("Sum"));
+        Assert.Equal(32, activityExecutionRecord.Outputs!.GetValue<int>("Product"));
+
+        var activityOutputRegister = workflowFinishedRecord.WorkflowExecutionContext.GetActivityOutputRegister();
+        Assert.Equal(2, activityOutputRegister.FindMany(p => true).Count());
+        Assert.Equal(12, activityOutputRegister.FindOutputByActivityId("SampleActivity1", "Sum"));
+        Assert.Equal(32, activityOutputRegister.FindOutputByActivityId("SampleActivity1", "Product"));
+    }
+
+    [Theory(DisplayName = "Activity outputs captured in activity execution record")]
+    [InlineData(true)]
+    [InlineData(false)]
+    public async Task ActivityOutputCaptureParallelTest(bool runAsynchronously)
+    {
+
+        // Arrange
+
+        TestWorkflow workflow = new(workflowBuilder =>
+            {
+                workflowBuilder.DefinitionId = "sampleWorkflow";
+                workflowBuilder.Id = "sampleWorkflow";
+
+                var variable1 = new Variable<int>();
+                var variable2 = new Variable<int>();
+                workflowBuilder.Root = new Elsa.Workflows.Activities.Parallel()
+                {
+                    Variables =
+                        {
+                            variable1,
+                            variable2,
+                        },
+                    Activities =
+                        {
+                            new SampleActivity()
+                            {
+                                Id = "SampleActivity1",
+                                RunAsynchronously = runAsynchronously,
+                                Number1 = new(4),
+                                Number2 = new(8),
+                                Sum = new(variable1),
+                            },
+                            new SampleActivity()
+                            {
+                                Id = "SampleActivity2",
+                                RunAsynchronously = runAsynchronously,
+                                Number1 = new(2),
+                                Number2 = new(7),
+                                Product = new(variable2),
+                            }
+                        }
+                };
+            }
+        );
+
+        var activityExecutionStore = new MemoryActivityExecutionStore(new MemoryStore<ActivityExecutionRecord>());
+
+        // Act
+
+        WorkflowFinished? workflowFinishedRecord = await workflow.DispatchWorkflowAndRunToCompletion(
+            configureServices: services =>
+            {
+                services.AddScoped<DistributedWorkflowRuntime>();
+            },
+            configureElsa: elsa =>
+            {
+                elsa.UseWorkflowRuntime(workflowRuntime => {
+                    workflowRuntime.ActivityExecutionLogStore = sp => activityExecutionStore;
+                    workflowRuntime.WorkflowRuntime = sp => sp.GetRequiredService<DistributedWorkflowRuntime>();
+                });
+            }
+        );
+
+        // Assert
+
+        Assert.NotNull(workflowFinishedRecord);
+        Assert.Equal(WorkflowStatus.Finished, workflowFinishedRecord.WorkflowState.Status);
+        Assert.Equal(WorkflowSubStatus.Finished, workflowFinishedRecord.WorkflowState.SubStatus);
+
+        var activityExecutionRecord1 = await activityExecutionStore.FindAsync(new() { ActivityId = "SampleActivity1" });
+        Assert.NotNull(activityExecutionRecord1?.Outputs);
+        Assert.Equal(2, activityExecutionRecord1.Outputs!.Count);
+        Assert.Equal(12, activityExecutionRecord1.Outputs!.GetValue<int>("Sum"));
+        Assert.Equal(32, activityExecutionRecord1.Outputs!.GetValue<int>("Product"));
+
+        var activityExecutionRecord2 = await activityExecutionStore.FindAsync(new() { ActivityId = "SampleActivity2" });
+        Assert.NotNull(activityExecutionRecord2?.Outputs);
+        Assert.Equal(2, activityExecutionRecord2.Outputs!.Count);
+        Assert.Equal(9, activityExecutionRecord2.Outputs!.GetValue<int>("Sum"));
+        Assert.Equal(14, activityExecutionRecord2.Outputs!.GetValue<int>("Product"));
+    }
+}