SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

fix(runtime): preserve bookmark identity on unmatched-resume enqueue (#8154)

Browse source 
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 <cursoragent@cursor.com>
This commit is contained in:
Sipke Schoorstra 2026-09-14 15:29:07 +02:00 committed by GitHub
parent ed94f88b01
commit 3c4549752d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 139 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
src/modules/Elsa.Workflows.Runtime/Services/StimulusSender.cs
View file

@ -19,11 +19,16 @@ public class StimulusSender(
public Task<SendStimulusResult> 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);
}
/// <inheritdoc />
public async Task<SendStimulusResult> SendAsync(string stimulusHash, StimulusMetadata? metadata = null, CancellationToken cancellationToken = default)
public Task<SendStimulusResult> SendAsync(string stimulusHash, StimulusMetadata? metadata = null, CancellationToken cancellationToken = default)
{
return SendAsync(stimulusHash, metadata, activityTypeName: null, cancellationToken);
}
private async Task<SendStimulusResult> SendAsync(string stimulusHash, StimulusMetadata? metadata, string? activityTypeName, CancellationToken cancellationToken)
{
var responses = new List<RunWorkflowInstanceResponse>();
@ -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<ICollection<RunWorkflowInstanceResponse>> ResumeExistingWorkflowsAsync(string stimulusHash, StimulusMetadata? metadata, CancellationToken cancellationToken)
private async Task<ICollection<RunWorkflowInstanceResponse>> 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,

128
test/unit/Elsa.Workflows.Runtime.UnitTests/Services/StimulusSenderTests.cs Normal file
View file

@ -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<IStimulusHasher>();
private readonly ITriggerBoundWorkflowService _triggerBoundWorkflowService = Substitute.For<ITriggerBoundWorkflowService>();
private readonly IWorkflowResumer _workflowResumer = Substitute.For<IWorkflowResumer>();
private readonly IBookmarkQueue _bookmarkQueue = Substitute.For<IBookmarkQueue>();
private readonly ITriggerInvoker _triggerInvoker = Substitute.For<ITriggerInvoker>();
public StimulusSenderTests()
{
_stimulusHasher.Hash(ActivityTypeName, Arg.Any<object>(), ActivityInstanceId).Returns(StimulusHash);
_workflowResumer
.ResumeAsync(Arg.Any<BookmarkFilter>(), Arg.Any<ResumeBookmarkOptions>(), Arg.Any<CancellationToken>())
.Returns([]);
_triggerBoundWorkflowService
.FindManyAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
.Returns([]);
}
[Fact]
public async Task SendAsync_WhenUnmatchedResume_EnqueuesActivityInstanceIdAndActivityTypeName()
{
var sender = CreateSender();
var metadata = CreateMetadata();
NewBookmarkQueueItem? enqueued = null;
await _bookmarkQueue.EnqueueAsync(Arg.Do<NewBookmarkQueueItem>(item => enqueued = item), Arg.Any<CancellationToken>());
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<NewBookmarkQueueItem>(item => enqueued = item), Arg.Any<CancellationToken>());
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<BookmarkFilter>(), Arg.Any<ResumeBookmarkOptions>(), Arg.Any<CancellationToken>())
.Returns([new RunWorkflowInstanceResponse { WorkflowInstanceId = WorkflowInstanceId }]);
var sender = CreateSender();
await sender.SendAsync(ActivityTypeName, new object(), CreateMetadata());
await _bookmarkQueue.DidNotReceive().EnqueueAsync(Arg.Any<NewBookmarkQueueItem>(), Arg.Any<CancellationToken>());
}
private StimulusSender CreateSender()
{
return new(
_stimulusHasher,
_triggerBoundWorkflowService,
_workflowResumer,
_bookmarkQueue,
_triggerInvoker,
NullLogger<StimulusSender>.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();
}
}