diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs index 7149728fdd..a083a260ef 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs +++ b/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 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 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); } -} \ No newline at end of file +} diff --git a/src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs index b217082f54..f2ae0cf986 100644 --- a/src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Options/BookmarkQueuePurgeOptions.cs @@ -8,10 +8,10 @@ public class BookmarkQueuePurgeOptions /// /// The time-to-live for bookmark queue items. /// - public TimeSpan Ttl { get; set; } = TimeSpan.FromMinutes(1); + public TimeSpan Ttl { get; set; } = TimeSpan.FromHours(1); /// /// The number of records to clean up per sweep. /// public int BatchSize { get; set; } = 1000; -} \ No newline at end of file +} diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs index 7035d569ea..f7b7438177 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs +++ b/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 _logger; + + protected IBookmarkQueueSignaler Signaler => _signaler; + protected ILogger Logger => _logger; public BookmarkQueueWorker(IBookmarkQueueSignaler signaler, IServiceScopeFactory scopeFactory, ILogger logger) { @@ -71,4 +74,4 @@ protected virtual async Task ProcessAsync(CancellationToken cancellationToken) await processor.ProcessAsync(cancellationToken); _logger.LogDebug("Processed bookmark queue."); } -} \ No newline at end of file +} diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Elsa.Workflows.Runtime.UnitTests.csproj b/test/unit/Elsa.Workflows.Runtime.UnitTests/Elsa.Workflows.Runtime.UnitTests.csproj index 3a3ee2d175..aa1ff85c95 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Elsa.Workflows.Runtime.UnitTests.csproj +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Elsa.Workflows.Runtime.UnitTests.csproj @@ -6,6 +6,7 @@ + diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Options/BookmarkQueuePurgeOptionsTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Options/BookmarkQueuePurgeOptionsTests.cs new file mode 100644 index 0000000000..ed3337384a --- /dev/null +++ b/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); + } +} diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DistributedBookmarkQueueWorkerTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DistributedBookmarkQueueWorkerTests.cs new file mode 100644 index 0000000000..2313f4e84f --- /dev/null +++ b/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(); + var scopeFactory = Substitute.For(); + var logger = Substitute.For>(); + var worker = new TestDistributedBookmarkQueueWorker(distributedLockProvider, signaler, scopeFactory, logger); + + await worker.InvokeProcessAsync(CancellationToken.None); + + await signaler.Received(1).TriggerAsync(Arg.Any()); + scopeFactory.DidNotReceive().CreateScope(); + } + + private class TestDistributedBookmarkQueueWorker( + IDistributedLockProvider distributedLockProvider, + IBookmarkQueueSignaler signaler, + IServiceScopeFactory scopeFactory, + ILogger 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 AcquireAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) + { + throw new NotSupportedException(); + } + + public IDistributedSynchronizationHandle? TryAcquire(TimeSpan timeout = default, CancellationToken cancellationToken = default) => null; + + public ValueTask TryAcquireAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => ValueTask.FromResult(null); + } +}