From 942cca32a150639f332e5c6862ecfa54112eda6c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 25 Apr 2026 12:37:45 +0000 Subject: [PATCH 1/3] Initial plan From 6cd8018e7aa08c9ad883bf80f6fd91452d1cca65 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 25 Apr 2026 12:42:10 +0000 Subject: [PATCH 2/3] Fix distributed bookmark queue retry race Agent-Logs-Url: https://github.com/elsa-workflows/elsa-core/sessions/a518907b-9fca-460b-a81c-fff600642fa2 Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --- .../DistributedBookmarkQueueWorker.cs | 30 +++++++--- .../Options/BookmarkQueuePurgeOptions.cs | 4 +- .../Elsa.Workflows.Runtime.UnitTests.csproj | 1 + .../Options/BookmarkQueuePurgeOptionsTests.cs | 14 +++++ .../DistributedBookmarkQueueWorkerTests.cs | 58 +++++++++++++++++++ 5 files changed, 97 insertions(+), 10 deletions(-) create mode 100644 test/unit/Elsa.Workflows.Runtime.UnitTests/Options/BookmarkQueuePurgeOptionsTests.cs create mode 100644 test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DistributedBookmarkQueueWorkerTests.cs diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs index 7149728fdd..b6fbdf934e 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs @@ -4,22 +4,36 @@ 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 LockUnavailableRetryDelay = TimeSpan.FromSeconds(1); + private readonly IDistributedLockProvider _distributedLockProvider; + private readonly IBookmarkQueueSignaler _signaler; + private readonly ILogger _logger; + + public DistributedBookmarkQueueWorker( + IDistributedLockProvider distributedLockProvider, + IBookmarkQueueSignaler signaler, + IServiceScopeFactory scopeFactory, + ILogger logger) : base(signaler, scopeFactory, logger) + { + _distributedLockProvider = distributedLockProvider; + _signaler = signaler; + _logger = logger; + } + 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(LockUnavailableRetryDelay, 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/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..b11cbc1621 --- /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.ProcessOnceAsync(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 ProcessOnceAsync(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); + } +} From 167eba790150c95a2453c46198d8445d1211b69c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 25 Apr 2026 12:46:23 +0000 Subject: [PATCH 3/3] Address bookmark queue review feedback Agent-Logs-Url: https://github.com/elsa-workflows/elsa-core/sessions/a518907b-9fca-460b-a81c-fff600642fa2 Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --- .../Services/DistributedBookmarkQueueWorker.cs | 12 ++++-------- .../Services/BookmarkQueueWorker.cs | 5 ++++- .../Services/DistributedBookmarkQueueWorkerTests.cs | 4 ++-- 3 files changed, 10 insertions(+), 11 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs index b6fbdf934e..a083a260ef 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedBookmarkQueueWorker.cs @@ -6,10 +6,8 @@ namespace Elsa.Workflows.Runtime.Distributed; public class DistributedBookmarkQueueWorker : BookmarkQueueWorker { - private static readonly TimeSpan LockUnavailableRetryDelay = TimeSpan.FromSeconds(1); + private static readonly TimeSpan RetryDelay = TimeSpan.FromSeconds(1); private readonly IDistributedLockProvider _distributedLockProvider; - private readonly IBookmarkQueueSignaler _signaler; - private readonly ILogger _logger; public DistributedBookmarkQueueWorker( IDistributedLockProvider distributedLockProvider, @@ -18,8 +16,6 @@ public DistributedBookmarkQueueWorker( ILogger logger) : base(signaler, scopeFactory, logger) { _distributedLockProvider = distributedLockProvider; - _signaler = signaler; - _logger = logger; } protected override async Task ProcessAsync(CancellationToken cancellationToken) @@ -28,9 +24,9 @@ protected override async Task ProcessAsync(CancellationToken cancellationToken) if (handle == null) { - _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(LockUnavailableRetryDelay, cancellationToken); - await _signaler.TriggerAsync(cancellationToken); + 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; } 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/Services/DistributedBookmarkQueueWorkerTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DistributedBookmarkQueueWorkerTests.cs index b11cbc1621..2313f4e84f 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DistributedBookmarkQueueWorkerTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DistributedBookmarkQueueWorkerTests.cs @@ -17,7 +17,7 @@ public async Task ProcessAsync_LockUnavailable_TriggersRetrySignal() var logger = Substitute.For>(); var worker = new TestDistributedBookmarkQueueWorker(distributedLockProvider, signaler, scopeFactory, logger); - await worker.ProcessOnceAsync(CancellationToken.None); + await worker.InvokeProcessAsync(CancellationToken.None); await signaler.Received(1).TriggerAsync(Arg.Any()); scopeFactory.DidNotReceive().CreateScope(); @@ -29,7 +29,7 @@ private class TestDistributedBookmarkQueueWorker( IServiceScopeFactory scopeFactory, ILogger logger) : DistributedBookmarkQueueWorker(distributedLockProvider, signaler, scopeFactory, logger) { - public Task ProcessOnceAsync(CancellationToken cancellationToken) => ProcessAsync(cancellationToken); + public Task InvokeProcessAsync(CancellationToken cancellationToken) => ProcessAsync(cancellationToken); } private class UnavailableDistributedLockProvider : IDistributedLockProvider