Fix distributed bookmark queue resume race

src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs

@@ -4,22 +4,32 @@
 
 namespace Elsa.Workflows.Runtime.Distributed;
 
-public class DistributedBookmarkQueueWorker(
-    IDistributedLockProvider distributedLockProvider,
-    IBookmarkQueueSignaler signaler,
-    IServiceScopeFactory scopeFactory,
-    ILogger<DistributedBookmarkQueueWorker> logger) : BookmarkQueueWorker(signaler, scopeFactory, logger)
+public class DistributedBookmarkQueueWorker : BookmarkQueueWorker
 {
+    private static readonly TimeSpan RetryDelay = TimeSpan.FromSeconds(1);
+    private readonly IDistributedLockProvider _distributedLockProvider;
+
+    public DistributedBookmarkQueueWorker(
+        IDistributedLockProvider distributedLockProvider,
+        IBookmarkQueueSignaler signaler,
+        IServiceScopeFactory scopeFactory,
+        ILogger<DistributedBookmarkQueueWorker> logger) : base(signaler, scopeFactory, logger)
+    {
+        _distributedLockProvider = distributedLockProvider;
+    }
+
     protected override async Task ProcessAsync(CancellationToken cancellationToken)
     {
-        await using var handle = await distributedLockProvider.TryAcquireLockAsync(nameof(DistributedBookmarkQueueWorker), TimeSpan.Zero, cancellationToken);
+        await using var handle = await _distributedLockProvider.TryAcquireLockAsync(nameof(DistributedBookmarkQueueWorker), TimeSpan.Zero, cancellationToken);
 
         if (handle == null)
         {
-            logger.LogInformation("Could not acquire lock for distributed bookmark queue worker. This is usually an indication that another application instance is already processing.");
+            Logger.LogDebug("Could not acquire lock for distributed bookmark queue worker. This is usually an indication that another application instance is already processing.");
+            await Task.Delay(RetryDelay, cancellationToken);
+            await Signaler.TriggerAsync(cancellationToken);
             return;
         }
 
         await base.ProcessAsync(cancellationToken);
     }
-}
+}

src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs

@@ -8,10 +8,10 @@ public class BookmarkQueuePurgeOptions
     /// <summary>
     /// The time-to-live for bookmark queue items.
     /// </summary>
-    public TimeSpan Ttl { get; set; } = TimeSpan.FromMinutes(1);
+    public TimeSpan Ttl { get; set; } = TimeSpan.FromHours(1);
 
     /// <summary>
     /// The number of records to clean up per sweep.
     /// </summary>
     public int BatchSize { get; set; } = 1000;
-}
+}

src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs

@@ -12,6 +12,9 @@ public class BookmarkQueueWorker : IBookmarkQueueWorker
     private readonly IBookmarkQueueSignaler _signaler;
     private readonly IServiceScopeFactory _scopeFactory;
     private readonly ILogger<BookmarkQueueWorker> _logger;
+
+    protected IBookmarkQueueSignaler Signaler => _signaler;
+    protected ILogger<BookmarkQueueWorker> Logger => _logger;
 
     public BookmarkQueueWorker(IBookmarkQueueSignaler signaler, IServiceScopeFactory scopeFactory, ILogger<BookmarkQueueWorker> logger)
     {
@@ -71,4 +74,4 @@ protected virtual async Task ProcessAsync(CancellationToken cancellationToken)
         await processor.ProcessAsync(cancellationToken);
         _logger.LogDebug("Processed bookmark queue.");
     }
-}
+}

test/unit/Elsa.Workflows.Runtime.UnitTests/Elsa.Workflows.Runtime.UnitTests.csproj

@@ -6,6 +6,7 @@
 
     <ItemGroup>
       <ProjectReference Include="..\..\..\src\modules\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
+      <ProjectReference Include="..\..\..\src\modules\Elsa.Workflows.Runtime.Distributed\Elsa.Workflows.Runtime.Distributed.csproj" />
       <ProjectReference Include="..\..\..\src\common\Elsa.Testing.Shared\Elsa.Testing.Shared.csproj" />
     </ItemGroup>
 

test/unit/Elsa.Workflows.Runtime.UnitTests/Options/BookmarkQueuePurgeOptionsTests.cs

@@ -0,0 +1,14 @@
+using Elsa.Workflows.Runtime.Options;
+
+namespace Elsa.Workflows.Runtime.UnitTests.Options;
+
+public class BookmarkQueuePurgeOptionsTests
+{
+    [Fact(DisplayName = "Default TTL keeps queue items long enough for periodic retries")]
+    public void DefaultTtl_KeepsQueueItemsLongEnoughForPeriodicRetries()
+    {
+        var options = new BookmarkQueuePurgeOptions();
+
+        Assert.Equal(TimeSpan.FromHours(1), options.Ttl);
+    }
+}

test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DistributedBookmarkQueueWorkerTests.cs

@@ -0,0 +1,58 @@
+using Elsa.Workflows.Runtime.Distributed;
+using Medallion.Threading;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using NSubstitute;
+
+namespace Elsa.Workflows.Runtime.UnitTests.Services;
+
+public class DistributedBookmarkQueueWorkerTests
+{
+    [Fact(DisplayName = "ProcessAsync re-signals when distributed lock is unavailable")]
+    public async Task ProcessAsync_LockUnavailable_TriggersRetrySignal()
+    {
+        var distributedLockProvider = new UnavailableDistributedLockProvider();
+        var signaler = Substitute.For<IBookmarkQueueSignaler>();
+        var scopeFactory = Substitute.For<IServiceScopeFactory>();
+        var logger = Substitute.For<ILogger<DistributedBookmarkQueueWorker>>();
+        var worker = new TestDistributedBookmarkQueueWorker(distributedLockProvider, signaler, scopeFactory, logger);
+
+        await worker.InvokeProcessAsync(CancellationToken.None);
+
+        await signaler.Received(1).TriggerAsync(Arg.Any<CancellationToken>());
+        scopeFactory.DidNotReceive().CreateScope();
+    }
+
+    private class TestDistributedBookmarkQueueWorker(
+        IDistributedLockProvider distributedLockProvider,
+        IBookmarkQueueSignaler signaler,
+        IServiceScopeFactory scopeFactory,
+        ILogger<DistributedBookmarkQueueWorker> logger) : DistributedBookmarkQueueWorker(distributedLockProvider, signaler, scopeFactory, logger)
+    {
+        public Task InvokeProcessAsync(CancellationToken cancellationToken) => ProcessAsync(cancellationToken);
+    }
+
+    private class UnavailableDistributedLockProvider : IDistributedLockProvider
+    {
+        public IDistributedLock CreateLock(string name) => new UnavailableDistributedLock(name);
+    }
+
+    private class UnavailableDistributedLock(string name) : IDistributedLock
+    {
+        public string Name { get; } = name;
+
+        public IDistributedSynchronizationHandle Acquire(TimeSpan? timeout = null, CancellationToken cancellationToken = default)
+        {
+            throw new NotSupportedException();
+        }
+
+        public ValueTask<IDistributedSynchronizationHandle> AcquireAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default)
+        {
+            throw new NotSupportedException();
+        }
+
+        public IDistributedSynchronizationHandle? TryAcquire(TimeSpan timeout = default, CancellationToken cancellationToken = default) => null;
+
+        public ValueTask<IDistributedSynchronizationHandle?> TryAcquireAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => ValueTask.FromResult<IDistributedSynchronizationHandle?>(null);
+    }
+}