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

fix: restore commit notification scope before flush

Browse source 
Detach the AsyncLocal scope synchronously so notifications published after an asynchronous flush are not re-buffered and discarded. Add regression coverage that forces the flush across an async boundary.
This commit is contained in:
Sipke Schoorstra 2026-07-31 04:50:31 +02:00
parent d63dae95bc
commit c3c6f12858
No known key found for this signature in database
GPG key ID: 5C10502B28A4268F
2 changed files with 36 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/modules/Elsa.Workflows.Runtime/Services/WorkflowCommitNotificationBuffer.cs
View file

@ -38,10 +38,15 @@ public class WorkflowCommitNotificationBuffer(IMediator mediator, ILogger<Workfl
_entries.Add(new(notification, strategy));
}
public async Task FlushAsync(CancellationToken cancellationToken = default)
public Task FlushAsync(CancellationToken cancellationToken = default)
{
ThrowIfDisposed();
owner._currentScope.Value = parent;
return FlushEntriesAsync(cancellationToken);
}
private async Task FlushEntriesAsync(CancellationToken cancellationToken)
{
List<Exception>? exceptions = null;
foreach (var entry in _entries)

30
test/unit/Elsa.Workflows.Runtime.UnitTests/Services/WorkflowCommitNotificationBufferTests.cs
View file

@ -24,6 +24,36 @@ public class WorkflowCommitNotificationBufferTests
await mediator.Received(1).SendAsync(notification, Arg.Any<IEventPublishingStrategy?>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task SendAsync_AfterScopeIsFlushed_PublishesImmediately()
{
var mediator = Substitute.For<IMediator>();
var buffer = CreateBuffer(mediator);
var sender = new WorkflowCommitNotificationSender(mediator, buffer);
var bufferedNotification = new TestNotification();
var subsequentNotification = new TestNotification();
var notificationPublishingStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var continueNotificationPublishing = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
mediator
.SendAsync(bufferedNotification, Arg.Any<IEventPublishingStrategy?>(), Arg.Any<CancellationToken>())
.Returns(async _ =>
{
notificationPublishingStarted.SetResult();
await continueNotificationPublishing.Task;
});
using var scope = buffer.Begin();
await sender.SendAsync(bufferedNotification);
var flushTask = scope.FlushAsync();
await notificationPublishingStarted.Task;
continueNotificationPublishing.SetResult();
await flushTask;
await sender.SendAsync(subsequentNotification);
await mediator.Received(1).SendAsync(bufferedNotification, Arg.Any<IEventPublishingStrategy?>(), Arg.Any<CancellationToken>());
await mediator.Received(1).SendAsync(subsequentNotification, Arg.Any<IEventPublishingStrategy?>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task SendAsync_WhenBufferingScopeIsDisposedWithoutFlush_DiscardsNotifications()
{