From 97c62a718cea0407a99bad191454b62cb2eb43e6 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 23 Jun 2026 03:07:50 +0200 Subject: [PATCH] Address PR review comments Document why activity execution context taint is cleared after commit instead of in the log sink, and refactor commit handler tests to share repeated fixture setup. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Services/StoreActivityExecutionLogSink.cs | 1 + .../DefaultCommitStateHandlerTests.cs | 170 ++++++++---------- 2 files changed, 73 insertions(+), 98 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs b/src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs index 99d0a1c8d..580b26e07 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs @@ -24,6 +24,7 @@ public class StoreActivityExecutionLogSink( var records = await Task.WhenAll(activityExecutionContexts.Select(x => x.GetOrMapCapturedActivityExecutionRecordAsync())); await activityExecutionStore.SaveManyAsync(records, cancellationToken); + // DefaultCommitStateHandler clears taint after the commit transaction succeeds. await notificationSender.SendAsync(new ActivityExecutionLogUpdated(context, records), cancellationToken); } } \ No newline at end of file diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DefaultCommitStateHandlerTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DefaultCommitStateHandlerTests.cs index 8bda422bc..5c51bc838 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DefaultCommitStateHandlerTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DefaultCommitStateHandlerTests.cs @@ -17,129 +17,103 @@ public class DefaultCommitStateHandlerTests [Fact] public async Task CommitAsync_ExecutesPersistenceInsideCommitTransaction() { - var (workflowExecutionContext, dirtyActivityExecutionContext) = await CreateWorkflowExecutionContextAsync(); - dirtyActivityExecutionContext.Taint(); - var workflowState = new WorkflowState { Id = workflowExecutionContext.Id }; - var workflowInstance = new WorkflowInstance { Id = workflowExecutionContext.Id }; - var workflowInstanceManager = Substitute.For(); - var bookmarkPersister = Substitute.For(); - var variablePersistenceManager = Substitute.For(); - var mediator = Substitute.For(); - var notificationBuffer = CreateBuffer(mediator); - var notificationSender = Substitute.For(); - var activityExecutionLogSink = Substitute.For>(); - var workflowExecutionLogSink = Substitute.For>(); - var transaction = new RecordingWorkflowCommitTransaction(); - workflowInstanceManager.SaveAsync(workflowState, Arg.Any()).Returns(workflowInstance); - bookmarkPersister.When(x => x.PersistBookmarksAsync(Arg.Any())).Do(_ => Assert.True(transaction.IsExecuting)); - activityExecutionLogSink.When(x => x.PersistExecutionLogsAsync(workflowExecutionContext, Arg.Any())).Do(_ => Assert.True(transaction.IsExecuting)); - workflowExecutionLogSink.When(x => x.PersistExecutionLogsAsync(workflowExecutionContext, Arg.Any())).Do(_ => Assert.True(transaction.IsExecuting)); - variablePersistenceManager.When(x => x.SaveVariablesAsync(workflowExecutionContext)).Do(_ => Assert.True(transaction.IsExecuting)); - workflowInstanceManager.When(x => x.SaveAsync(workflowState, Arg.Any())).Do(_ => Assert.True(transaction.IsExecuting)); - var handler = new DefaultCommitStateHandler( - workflowInstanceManager, - bookmarkPersister, - variablePersistenceManager, - transaction, - notificationBuffer, - notificationSender, - activityExecutionLogSink, - workflowExecutionLogSink); + var fixture = await CommitTestFixture.CreateAsync(); + fixture.WorkflowInstanceManager.SaveAsync(fixture.WorkflowState, Arg.Any()).Returns(fixture.WorkflowInstance); + fixture.BookmarkPersister.When(x => x.PersistBookmarksAsync(Arg.Any())).Do(_ => Assert.True(fixture.Transaction.IsExecuting)); + fixture.ActivityExecutionLogSink.When(x => x.PersistExecutionLogsAsync(fixture.WorkflowExecutionContext, Arg.Any())).Do(_ => Assert.True(fixture.Transaction.IsExecuting)); + fixture.WorkflowExecutionLogSink.When(x => x.PersistExecutionLogsAsync(fixture.WorkflowExecutionContext, Arg.Any())).Do(_ => Assert.True(fixture.Transaction.IsExecuting)); + fixture.VariablePersistenceManager.When(x => x.SaveVariablesAsync(fixture.WorkflowExecutionContext)).Do(_ => Assert.True(fixture.Transaction.IsExecuting)); + fixture.WorkflowInstanceManager.When(x => x.SaveAsync(fixture.WorkflowState, Arg.Any())).Do(_ => Assert.True(fixture.Transaction.IsExecuting)); - await handler.CommitAsync(workflowExecutionContext, workflowState); + await fixture.Handler.CommitAsync(fixture.WorkflowExecutionContext, fixture.WorkflowState); - Assert.True(transaction.Completed); - Assert.False(dirtyActivityExecutionContext.IsDirty); - await notificationSender.Received(1).SendAsync( - Arg.Is(x => x.WorkflowInstance == workflowInstance && x.WorkflowState == workflowState), + Assert.True(fixture.Transaction.Completed); + Assert.False(fixture.ActivityExecutionContext.IsDirty); + await fixture.NotificationSender.Received(1).SendAsync( + Arg.Is(x => x.WorkflowInstance == fixture.WorkflowInstance && x.WorkflowState == fixture.WorkflowState), Arg.Any()); } [Fact] public async Task CommitAsync_WhenPersistenceFails_DoesNotClearExecutionLogOrPublishCommittedNotification() { - var (workflowExecutionContext, dirtyActivityExecutionContext) = await CreateWorkflowExecutionContextAsync(); - dirtyActivityExecutionContext.Taint(); - workflowExecutionContext.AddExecutionLogEntry("Started"); - var workflowState = new WorkflowState { Id = workflowExecutionContext.Id }; - var workflowInstanceManager = Substitute.For(); - var bookmarkPersister = Substitute.For(); - var variablePersistenceManager = Substitute.For(); - var mediator = Substitute.For(); - var notificationBuffer = CreateBuffer(mediator); - var notificationSender = Substitute.For(); - var activityExecutionLogSink = Substitute.For>(); - var workflowExecutionLogSink = Substitute.For>(); - var transaction = new RecordingWorkflowCommitTransaction(); - workflowInstanceManager.SaveAsync(workflowState, Arg.Any()).Returns>(_ => throw new InvalidOperationException("state save failed")); - var handler = new DefaultCommitStateHandler( - workflowInstanceManager, - bookmarkPersister, - variablePersistenceManager, - transaction, - notificationBuffer, - notificationSender, - activityExecutionLogSink, - workflowExecutionLogSink); + var fixture = await CommitTestFixture.CreateAsync(); + fixture.WorkflowExecutionContext.AddExecutionLogEntry("Started"); + fixture.WorkflowInstanceManager.SaveAsync(fixture.WorkflowState, Arg.Any()).Returns>(_ => throw new InvalidOperationException("state save failed")); - await Assert.ThrowsAsync(() => handler.CommitAsync(workflowExecutionContext, workflowState)); + await Assert.ThrowsAsync(() => fixture.Handler.CommitAsync(fixture.WorkflowExecutionContext, fixture.WorkflowState)); - Assert.True(transaction.Executed); - Assert.False(transaction.Completed); - Assert.True(dirtyActivityExecutionContext.IsDirty); - Assert.NotEmpty(workflowExecutionContext.ExecutionLog); - await notificationSender.DidNotReceive().SendAsync(Arg.Any(), Arg.Any()); + Assert.True(fixture.Transaction.Executed); + Assert.False(fixture.Transaction.Completed); + Assert.True(fixture.ActivityExecutionContext.IsDirty); + Assert.NotEmpty(fixture.WorkflowExecutionContext.ExecutionLog); + await fixture.NotificationSender.DidNotReceive().SendAsync(Arg.Any(), Arg.Any()); } [Fact] public async Task CommitAsync_WhenBufferedNotificationFlushFails_PublishesCommittedNotification() { - var (workflowExecutionContext, _) = await CreateWorkflowExecutionContextAsync(); - var workflowState = new WorkflowState { Id = workflowExecutionContext.Id }; - var workflowInstance = new WorkflowInstance { Id = workflowExecutionContext.Id }; - var workflowInstanceManager = Substitute.For(); - var bookmarkPersister = Substitute.For(); - var variablePersistenceManager = Substitute.For(); - var mediator = Substitute.For(); - var notificationBuffer = CreateBuffer(mediator); - var bufferedNotificationSender = new WorkflowCommitNotificationSender(mediator, notificationBuffer); - var notificationSender = Substitute.For(); - var activityExecutionLogSink = Substitute.For>(); - var workflowExecutionLogSink = Substitute.For>(); - var transaction = new RecordingWorkflowCommitTransaction(); + var fixture = await CommitTestFixture.CreateAsync(); + var bufferedNotificationSender = new WorkflowCommitNotificationSender(fixture.Mediator, fixture.NotificationBuffer); var bufferedNotification = new TestNotification(); - workflowInstanceManager.SaveAsync(workflowState, Arg.Any()).Returns(workflowInstance); - activityExecutionLogSink - .PersistExecutionLogsAsync(workflowExecutionContext, Arg.Any()) + fixture.WorkflowInstanceManager.SaveAsync(fixture.WorkflowState, Arg.Any()).Returns(fixture.WorkflowInstance); + fixture.ActivityExecutionLogSink + .PersistExecutionLogsAsync(fixture.WorkflowExecutionContext, Arg.Any()) .Returns(_ => bufferedNotificationSender.SendAsync(bufferedNotification)); - mediator + fixture.Mediator .SendAsync(bufferedNotification, Arg.Any(), Arg.Any()) .Returns(_ => throw new InvalidOperationException("Buffered handler failed")); - var handler = new DefaultCommitStateHandler( - workflowInstanceManager, - bookmarkPersister, - variablePersistenceManager, - transaction, - notificationBuffer, - notificationSender, - activityExecutionLogSink, - workflowExecutionLogSink); - await Assert.ThrowsAsync(() => handler.CommitAsync(workflowExecutionContext, workflowState)); + await Assert.ThrowsAsync(() => fixture.Handler.CommitAsync(fixture.WorkflowExecutionContext, fixture.WorkflowState)); - await notificationSender.Received(1).SendAsync( - Arg.Is(x => x.WorkflowInstance == workflowInstance && x.WorkflowState == workflowState), + await fixture.NotificationSender.Received(1).SendAsync( + Arg.Is(x => x.WorkflowInstance == fixture.WorkflowInstance && x.WorkflowState == fixture.WorkflowState), Arg.Any()); } - private static async Task<(WorkflowExecutionContext WorkflowExecutionContext, ActivityExecutionContext DirtyActivityExecutionContext)> CreateWorkflowExecutionContextAsync() + private class CommitTestFixture { - var fixture = new ActivityTestFixture(new WriteLine("Test")); - var activityExecutionContext = await fixture.BuildAsync(); - activityExecutionContext.Taint(); - var workflowExecutionContext = activityExecutionContext.WorkflowExecutionContext; - workflowExecutionContext.AddActivityExecutionContext(activityExecutionContext); - return (workflowExecutionContext, activityExecutionContext); + private CommitTestFixture(ActivityExecutionContext activityExecutionContext) + { + ActivityExecutionContext = activityExecutionContext; + ActivityExecutionContext.Taint(); + WorkflowExecutionContext = activityExecutionContext.WorkflowExecutionContext; + WorkflowExecutionContext.AddActivityExecutionContext(activityExecutionContext); + WorkflowState = new() { Id = WorkflowExecutionContext.Id }; + WorkflowInstance = new() { Id = WorkflowExecutionContext.Id }; + NotificationBuffer = CreateBuffer(Mediator); + Handler = new( + WorkflowInstanceManager, + BookmarkPersister, + VariablePersistenceManager, + Transaction, + NotificationBuffer, + NotificationSender, + ActivityExecutionLogSink, + WorkflowExecutionLogSink); + } + + public WorkflowExecutionContext WorkflowExecutionContext { get; } + public ActivityExecutionContext ActivityExecutionContext { get; } + public WorkflowState WorkflowState { get; } + public WorkflowInstance WorkflowInstance { get; } + public IWorkflowInstanceManager WorkflowInstanceManager { get; } = Substitute.For(); + public IBookmarksPersister BookmarkPersister { get; } = Substitute.For(); + public IVariablePersistenceManager VariablePersistenceManager { get; } = Substitute.For(); + public IMediator Mediator { get; } = Substitute.For(); + public WorkflowCommitNotificationBuffer NotificationBuffer { get; } + public INotificationSender NotificationSender { get; } = Substitute.For(); + public ILogRecordSink ActivityExecutionLogSink { get; } = Substitute.For>(); + public ILogRecordSink WorkflowExecutionLogSink { get; } = Substitute.For>(); + public RecordingWorkflowCommitTransaction Transaction { get; } = new(); + public DefaultCommitStateHandler Handler { get; } + + public static async Task CreateAsync() + { + var fixture = new ActivityTestFixture(new WriteLine("Test")); + var activityExecutionContext = await fixture.BuildAsync(); + return new(activityExecutionContext); + } } private static WorkflowCommitNotificationBuffer CreateBuffer(IMediator mediator) => new(mediator, Substitute.For>());