Introduce TestBookmarkQueueWorker to eliminate throttling in component tests and ensure timely completion of workflows. Enhance disposal logic to prevent TaskCanceledException by waiting for workflows to complete.

test/component/Elsa.Workflows.ComponentTests/Helpers/Abstractions/AppComponentTest.cs

@@ -1,5 +1,8 @@
 using Elsa.Common.Multitenancy;
 using Elsa.Workflows.ComponentTests.Fixtures;
+using Elsa.Workflows.Management;
+using Elsa.Workflows.Management.Filters;
+using Elsa.Workflows.Models;
 using Microsoft.Extensions.DependencyInjection;
 
 namespace Elsa.Workflows.ComponentTests.Abstractions;
@@ -26,6 +29,10 @@ public abstract class AppComponentTest : IDisposable
 
     void IDisposable.Dispose()
     {
+        // Wait for all workflows to reach terminal state before disposing scope
+        // This prevents TaskCanceledException when workflows are still executing
+        WaitForWorkflowsToComplete();
+
         _tenantScope.Dispose();
         Scope.Dispose();
         OnDispose();
@@ -34,4 +41,40 @@ public abstract class AppComponentTest : IDisposable
     protected virtual void OnDispose()
     {
     }
+
+    private void WaitForWorkflowsToComplete()
+    {
+        try
+        {
+            var workflowInstanceStore = Scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
+            var timeout = TimeSpan.FromSeconds(10);
+            var pollInterval = TimeSpan.FromMilliseconds(50);
+            var deadline = DateTime.UtcNow.Add(timeout);
+
+            while (DateTime.UtcNow < deadline)
+            {
+                var filter = new WorkflowInstanceFilter
+                {
+                    WorkflowStatus = WorkflowStatus.Running
+                };
+
+                // Use async method synchronously - acceptable in cleanup/dispose
+                var runningWorkflows = workflowInstanceStore.FindManyAsync(filter, CancellationToken.None)
+                    .GetAwaiter()
+                    .GetResult();
+
+                if (!runningWorkflows.Any())
+                    return; // All workflows completed
+
+                Thread.Sleep(pollInterval);
+            }
+
+            // If we reach here, workflows didn't complete in time
+            // Log but don't throw to avoid masking actual test failures
+        }
+        catch
+        {
+            // Swallow exceptions during cleanup to avoid masking test failures
+        }
+    }
 }

test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs

@@ -109,6 +109,8 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl
                     });
                     runtime.UseCache();
                     runtime.UseDistributedRuntime();
+                    // Use test-specific bookmark queue worker without throttling to prevent timeouts
+                    runtime.BookmarkQueueWorker = sp => sp.GetRequiredService<TestBookmarkQueueWorker>();
                 });
                 elsa.UseJavaScript(options =>
                 {
@@ -168,6 +170,7 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl
                 .AddWorkflowsProvider<TestWorkflowProvider>()
                 .AddNotificationHandlersFrom<WorkflowEventHandlers>()
                 .Decorate<IChangeTokenSignaler, EventPublishingChangeTokenSignaler>()
+                .AddSingleton<TestBookmarkQueueWorker>()
                 ;
         });
     }

test/component/Elsa.Workflows.ComponentTests/Helpers/Services/TestBookmarkQueueWorker.cs

@@ -0,0 +1,67 @@
+using Elsa.Workflows.Runtime;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Workflows.ComponentTests.Services;
+
+/// <summary>
+/// A test-specific bookmark queue worker that processes items immediately without throttling.
+/// This prevents timeouts in tests where many workflows complete rapidly.
+/// </summary>
+public class TestBookmarkQueueWorker(IBookmarkQueueSignaler signaler, IServiceScopeFactory scopeFactory, ILogger<TestBookmarkQueueWorker> logger) : IBookmarkQueueWorker
+{
+    private CancellationTokenSource _cts = null!;
+    private bool _running;
+
+    public void Start()
+    {
+        if (_running)
+            return;
+
+        _cts = new();
+        _running = true;
+
+        _ = Task.Run(AwaitSignalAsync);
+    }
+
+    public void Stop()
+    {
+        if (_running)
+        {
+            _running = false;
+            _cts.Cancel();
+        }
+
+        _cts.Dispose();
+    }
+
+    private async Task AwaitSignalAsync()
+    {
+        while (!_cts.IsCancellationRequested)
+        {
+            try
+            {
+                await signaler.AwaitAsync(_cts.Token);
+                // Process immediately without throttling for tests
+                await ProcessAsync(_cts.Token);
+            }
+            catch (OperationCanceledException)
+            {
+                break; // Stop() was called
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, "TestBookmarkQueueWorker error – continuing loop");
+            }
+        }
+    }
+
+    protected virtual async Task ProcessAsync(CancellationToken cancellationToken)
+    {
+        logger.LogDebug("Processing bookmark queue (test mode - no throttling)...");
+        using var scope = scopeFactory.CreateScope();
+        var processor = scope.ServiceProvider.GetRequiredService<IBookmarkQueueProcessor>();
+        await processor.ProcessAsync(cancellationToken);
+        logger.LogDebug("Processed bookmark queue.");
+    }
+}