diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueProcessor.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueProcessor.cs index 341e0ef69d..8739869120 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueProcessor.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueProcessor.cs @@ -19,22 +19,33 @@ public async Task ProcessAsync(CancellationToken cancellationToken = default) var pageArgs = PageArgs.FromRange(offset, batchSize); var page = await store.PageAsync(pageArgs, new BookmarkQueueItemOrder(x => x.CreatedAt, OrderDirection.Ascending), cancellationToken); - await ProcessPageAsync(page, cancellationToken); + if (page.Items.Count == 0) + break; + + var deletedCount = await ProcessPageAsync(page, cancellationToken); + + // Advance past items that remain in the queue so they are retried on the next worker cycle, not reprocessed in this run. + offset += page.Items.Count - deletedCount; if (page.Items.Count < batchSize) break; - - offset += batchSize; } } - private async Task ProcessPageAsync(Page page, CancellationToken cancellationToken = default) + private async Task ProcessPageAsync(Page page, CancellationToken cancellationToken = default) { - foreach (var bookmarkQueueItem in page.Items) - await ProcessItemAsync(bookmarkQueueItem, cancellationToken); + var deletedCount = 0; + + foreach (var bookmarkQueueItem in page.Items) + { + if (await ProcessItemAsync(bookmarkQueueItem, cancellationToken)) + deletedCount++; + } + + return deletedCount; } - private async Task ProcessItemAsync(BookmarkQueueItem item, CancellationToken cancellationToken = default) + private async Task ProcessItemAsync(BookmarkQueueItem item, CancellationToken cancellationToken = default) { var filter = item.CreateBookmarkFilter(); var options = item.Options; @@ -47,10 +58,10 @@ private async Task ProcessItemAsync(BookmarkQueueItem item, CancellationToken ca { logger.LogDebug("Successfully resumed {WorkflowCount} workflow instances using stimulus {StimulusHash} for activity type {ActivityType}", responses.Count, item.StimulusHash, item.ActivityTypeName); await store.DeleteAsync(item.Id, cancellationToken); + return true; } - else - { - logger.LogDebug("No matching bookmarks found for bookmark queue item {BookmarkQueueItemId} for workflow instance {WorkflowInstanceId} for activity type {ActivityType} with stimulus {StimulusHash}", item.Id, item.WorkflowInstanceId, item.ActivityTypeName, item.StimulusHash); - } + + logger.LogDebug("No matching bookmarks found for bookmark queue item {BookmarkQueueItemId} for workflow instance {WorkflowInstanceId} for activity type {ActivityType} with stimulus {StimulusHash}", item.Id, item.WorkflowInstanceId, item.ActivityTypeName, item.StimulusHash); + return false; } } \ No newline at end of file diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/BookmarkQueueProcessorTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/BookmarkQueueProcessorTests.cs new file mode 100644 index 0000000000..ac8d7c165c --- /dev/null +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/BookmarkQueueProcessorTests.cs @@ -0,0 +1,113 @@ +using Elsa.Common.Services; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Messages; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Stores; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; + +namespace Elsa.Workflows.Runtime.UnitTests.Services; + +public class BookmarkQueueProcessorTests +{ + private readonly MemoryBookmarkQueueStore _store; + private readonly IWorkflowResumer _workflowResumer; + private readonly BookmarkQueueProcessor _processor; + private HashSet _failedBookmarkIds = []; + + public BookmarkQueueProcessorTests() + { + _store = new MemoryBookmarkQueueStore(new MemoryStore()); + _workflowResumer = Substitute.For(); + _workflowResumer + .ResumeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(callInfo => CreateResumeResult(callInfo.Arg())); + + _processor = new BookmarkQueueProcessor(_store, _workflowResumer, NullLogger.Instance); + } + + [Fact] + public async Task ProcessAsync_WhenAllItemsResumeSuccessfully_DrainsQueueLargerThanBatchSize() + { + await SeedQueueAsync(55); + + await _processor.ProcessAsync(); + + var remaining = (await _store.FindManyAsync(new BookmarkQueueFilter())).ToList(); + Assert.Empty(remaining); + } + + [Fact] + public async Task ProcessAsync_WhenSomeItemsFail_SkipsFailedItemsAndStillProcessesSubsequentPages() + { + await SeedQueueAsync(60, "item-05", "item-10"); + + await _processor.ProcessAsync(); + + var remaining = (await _store.FindManyAsync(new BookmarkQueueFilter())).Select(x => x.Id).OrderBy(x => x).ToList(); + Assert.Equal(["item-05", "item-10"], remaining); + } + + [Fact] + public async Task ProcessAsync_WhenRunAgain_RetriesPreviouslyFailedItems() + { + await SeedQueueAsync(60, "item-05", "item-10"); + + await _processor.ProcessAsync(); + + _failedBookmarkIds.Clear(); + + await _processor.ProcessAsync(); + + var remaining = (await _store.FindManyAsync(new BookmarkQueueFilter())).ToList(); + Assert.Empty(remaining); + } + + [Fact] + public async Task ProcessAsync_WhenEntirePageFails_AdvancesToNextPageWithoutReprocessingFailuresInSameRun() + { + var failingIds = Enumerable.Range(0, 50).Select(i => $"item-{i:D2}").ToArray(); + await SeedQueueAsync(55, failingIds); + _failedBookmarkIds = failingIds.ToHashSet(); + + await _processor.ProcessAsync(); + + var remaining = (await _store.FindManyAsync(new BookmarkQueueFilter())).Select(x => x.Id).OrderBy(x => x).ToList(); + Assert.Equal(Enumerable.Range(0, 50).Select(i => $"item-{i:D2}"), remaining); + } + + private async Task SeedQueueAsync(int count, params string[] failedIds) + { + _failedBookmarkIds = failedIds.ToHashSet(); + var baseTime = new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero); + + for (var index = 0; index < count; index++) + { + var id = $"item-{index:D2}"; + await _store.AddAsync(new BookmarkQueueItem + { + Id = id, + BookmarkId = id, + WorkflowInstanceId = $"wf-{index}", + ActivityTypeName = "TestActivity", + StimulusHash = $"hash-{index}", + CreatedAt = baseTime.AddSeconds(index), + }); + } + } + + private IEnumerable CreateResumeResult(BookmarkFilter filter) + { + if (filter.BookmarkId != null && _failedBookmarkIds.Contains(filter.BookmarkId)) + return []; + + return + [ + new RunWorkflowInstanceResponse + { + WorkflowInstanceId = filter.WorkflowInstanceId ?? "workflow-instance", + }, + ]; + } +}