fix: correct bookmark queue pagination after deletions

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<DateTimeOffset>(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<BookmarkQueueItem> page, CancellationToken cancellationToken = default)
+    private async Task<int> ProcessPageAsync(Page<BookmarkQueueItem> 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<bool> 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;
     }
 }

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<string> _failedBookmarkIds = [];
+
+    public BookmarkQueueProcessorTests()
+    {
+        _store = new MemoryBookmarkQueueStore(new MemoryStore<BookmarkQueueItem>());
+        _workflowResumer = Substitute.For<IWorkflowResumer>();
+        _workflowResumer
+            .ResumeAsync(Arg.Any<BookmarkFilter>(), Arg.Any<ResumeBookmarkOptions?>(), Arg.Any<CancellationToken>())
+            .Returns(callInfo => CreateResumeResult(callInfo.Arg<BookmarkFilter>()));
+
+        _processor = new BookmarkQueueProcessor(_store, _workflowResumer, NullLogger<BookmarkQueueProcessor>.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<RunWorkflowInstanceResponse> CreateResumeResult(BookmarkFilter filter)
+    {
+        if (filter.BookmarkId != null && _failedBookmarkIds.Contains(filter.BookmarkId))
+            return [];
+
+        return
+        [
+            new RunWorkflowInstanceResponse
+            {
+                WorkflowInstanceId = filter.WorkflowInstanceId ?? "workflow-instance",
+            },
+        ];
+    }
+}