From 3c4549752d89a2d86565bf6e9edd135ae65d8380 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 14 Sep 2026 15:29:07 +0200 Subject: [PATCH] fix(runtime): preserve bookmark identity on unmatched-resume enqueue (#8154) Copy ActivityInstanceId from metadata and ActivityTypeName from the typed SendAsync path onto NewBookmarkQueueItem so BookmarkQueueProcessor retries with the same filter identity as the live resume. Co-authored-by: Cursor Agent --- .../Services/StimulusSender.cs | 15 +- .../Services/StimulusSenderTests.cs | 128 ++++++++++++++++++ 2 files changed, 139 insertions(+), 4 deletions(-) create mode 100644 test/unit/Elsa.Workflows.Runtime.UnitTests/Services/StimulusSenderTests.cs diff --git a/src/modules/Elsa.Workflows.Runtime/Services/StimulusSender.cs b/src/modules/Elsa.Workflows.Runtime/Services/StimulusSender.cs index 2d93bb75b..de1bc4911 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/StimulusSender.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/StimulusSender.cs @@ -19,11 +19,16 @@ public class StimulusSender( public Task SendAsync(string activityTypeName, object stimulus, StimulusMetadata? metadata = null, CancellationToken cancellationToken = default) { var stimulusHash = stimulusHasher.Hash(activityTypeName, stimulus, metadata?.ActivityInstanceId); - return SendAsync(stimulusHash, metadata, cancellationToken); + return SendAsync(stimulusHash, metadata, activityTypeName, cancellationToken); } /// - public async Task SendAsync(string stimulusHash, StimulusMetadata? metadata = null, CancellationToken cancellationToken = default) + public Task SendAsync(string stimulusHash, StimulusMetadata? metadata = null, CancellationToken cancellationToken = default) + { + return SendAsync(stimulusHash, metadata, activityTypeName: null, cancellationToken); + } + + private async Task SendAsync(string stimulusHash, StimulusMetadata? metadata, string? activityTypeName, CancellationToken cancellationToken) { var responses = new List(); @@ -33,7 +38,7 @@ public class StimulusSender( responses.AddRange(triggered); } - var resumed = await ResumeExistingWorkflowsAsync(stimulusHash, metadata, cancellationToken); + var resumed = await ResumeExistingWorkflowsAsync(stimulusHash, metadata, activityTypeName, cancellationToken); responses.AddRange(resumed); return new(responses); } @@ -79,7 +84,7 @@ public class StimulusSender( return responses; } - private async Task> ResumeExistingWorkflowsAsync(string stimulusHash, StimulusMetadata? metadata, CancellationToken cancellationToken) + private async Task> ResumeExistingWorkflowsAsync(string stimulusHash, StimulusMetadata? metadata, string? activityTypeName, CancellationToken cancellationToken) { var input = metadata?.Input; var properties = metadata?.Properties; @@ -113,6 +118,8 @@ public class StimulusSender( BookmarkId = metadata?.BookmarkId, CorrelationId = metadata?.CorrelationId, StimulusHash = stimulusHash, + ActivityInstanceId = metadata?.ActivityInstanceId, + ActivityTypeName = activityTypeName, Options = new() { Input = input, diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/StimulusSenderTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/StimulusSenderTests.cs new file mode 100644 index 000000000..2c0cabdd3 --- /dev/null +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/StimulusSenderTests.cs @@ -0,0 +1,128 @@ +using Elsa.Workflows; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Messages; +using Elsa.Workflows.Runtime.Options; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; + +namespace Elsa.Workflows.Runtime.UnitTests.Services; + +public class StimulusSenderTests +{ + private const string ActivityTypeName = "Elsa.RunTask"; + private const string StimulusHash = "stimulus-hash"; + private const string ActivityInstanceId = "activity-instance-1"; + private const string WorkflowInstanceId = "workflow-instance-1"; + private const string BookmarkId = "bookmark-1"; + private const string CorrelationId = "correlation-1"; + + private readonly IStimulusHasher _stimulusHasher = Substitute.For(); + private readonly ITriggerBoundWorkflowService _triggerBoundWorkflowService = Substitute.For(); + private readonly IWorkflowResumer _workflowResumer = Substitute.For(); + private readonly IBookmarkQueue _bookmarkQueue = Substitute.For(); + private readonly ITriggerInvoker _triggerInvoker = Substitute.For(); + + public StimulusSenderTests() + { + _stimulusHasher.Hash(ActivityTypeName, Arg.Any(), ActivityInstanceId).Returns(StimulusHash); + _workflowResumer + .ResumeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns([]); + _triggerBoundWorkflowService + .FindManyAsync(Arg.Any(), Arg.Any()) + .Returns([]); + } + + [Fact] + public async Task SendAsync_WhenUnmatchedResume_EnqueuesActivityInstanceIdAndActivityTypeName() + { + var sender = CreateSender(); + var metadata = CreateMetadata(); + NewBookmarkQueueItem? enqueued = null; + await _bookmarkQueue.EnqueueAsync(Arg.Do(item => enqueued = item), Arg.Any()); + + await sender.SendAsync(ActivityTypeName, new object(), metadata); + + Assert.NotNull(enqueued); + Assert.Equal(ActivityInstanceId, enqueued.ActivityInstanceId); + Assert.Equal(ActivityTypeName, enqueued.ActivityTypeName); + Assert.Equal(WorkflowInstanceId, enqueued.WorkflowInstanceId); + Assert.Equal(BookmarkId, enqueued.BookmarkId); + Assert.Equal(CorrelationId, enqueued.CorrelationId); + Assert.Equal(StimulusHash, enqueued.StimulusHash); + + var filter = CreateBookmarkFilter(enqueued); + Assert.Equal(ActivityInstanceId, filter.ActivityInstanceId); + Assert.Equal(ActivityTypeName, filter.Name); + Assert.Equal(StimulusHash, filter.Hash); + } + + [Fact] + public async Task SendAsync_WhenUnmatchedResumeWithHashOnly_EnqueuesActivityInstanceIdWithoutActivityTypeName() + { + var sender = CreateSender(); + var metadata = CreateMetadata(); + NewBookmarkQueueItem? enqueued = null; + await _bookmarkQueue.EnqueueAsync(Arg.Do(item => enqueued = item), Arg.Any()); + + await sender.SendAsync(StimulusHash, metadata); + + Assert.NotNull(enqueued); + Assert.Equal(ActivityInstanceId, enqueued.ActivityInstanceId); + Assert.Null(enqueued.ActivityTypeName); + + var filter = CreateBookmarkFilter(enqueued); + Assert.Equal(ActivityInstanceId, filter.ActivityInstanceId); + Assert.Null(filter.Name); + } + + [Fact] + public async Task SendAsync_WhenResumeMatches_DoesNotEnqueue() + { + _workflowResumer + .ResumeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns([new RunWorkflowInstanceResponse { WorkflowInstanceId = WorkflowInstanceId }]); + var sender = CreateSender(); + + await sender.SendAsync(ActivityTypeName, new object(), CreateMetadata()); + + await _bookmarkQueue.DidNotReceive().EnqueueAsync(Arg.Any(), Arg.Any()); + } + + private StimulusSender CreateSender() + { + return new( + _stimulusHasher, + _triggerBoundWorkflowService, + _workflowResumer, + _bookmarkQueue, + _triggerInvoker, + NullLogger.Instance); + } + + private static StimulusMetadata CreateMetadata() + { + return new() + { + WorkflowInstanceId = WorkflowInstanceId, + ActivityInstanceId = ActivityInstanceId, + BookmarkId = BookmarkId, + CorrelationId = CorrelationId + }; + } + + private static BookmarkFilter CreateBookmarkFilter(NewBookmarkQueueItem item) + { + return new BookmarkQueueItem + { + WorkflowInstanceId = item.WorkflowInstanceId, + BookmarkId = item.BookmarkId, + CorrelationId = item.CorrelationId, + StimulusHash = item.StimulusHash, + ActivityInstanceId = item.ActivityInstanceId, + ActivityTypeName = item.ActivityTypeName, + Options = item.Options + }.CreateBookmarkFilter(); + } +}