From 5ba47b2a8aeaef9a8bdbc84d5955223c6a7c0c86 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 8 Sep 2026 17:03:55 +0000 Subject: [PATCH] Harden Redis queue uniqueness with atomic Lua enqueue. QueueWork used a non-atomic LINSERT/RPUSH/LREM sequence, so concurrent enqueues of a missing id could leave duplicate pending LIST entries. Keep the same LIST keys and algorithm, but run them in one EVAL so at most one pending occurrence exists per id. Dequeue stays a single LPOP. Co-authored-by: Daniel Gerlag --- .../Services/RedisQueueProvider.cs | 46 +++++- .../RedisQueueProviderFixture.cs | 155 ++++++++++++++++++ 2 files changed, 194 insertions(+), 7 deletions(-) create mode 100644 test/WorkflowCore.Tests.Redis/RedisQueueProviderFixture.cs diff --git a/src/providers/WorkflowCore.Providers.Redis/Services/RedisQueueProvider.cs b/src/providers/WorkflowCore.Providers.Redis/Services/RedisQueueProvider.cs index 347c699b6..2e6cddbea 100644 --- a/src/providers/WorkflowCore.Providers.Redis/Services/RedisQueueProvider.cs +++ b/src/providers/WorkflowCore.Providers.Redis/Services/RedisQueueProvider.cs @@ -8,8 +8,41 @@ namespace WorkflowCore.Providers.Redis.Services { + /// + /// Redis LIST-backed . Pending ids are unique per queue. + /// + /// + /// Compatibility: + /// + /// Uses the existing LIST keys only ({prefix}-workflows|events|index). No companion SET, + /// so there is no extra key to name, TTL, or clean up, and no pop/SREM crash window. + /// Existing LIST keys (including those that already contain duplicate entries) need no + /// migration or flush. Pre-existing duplicates drain via ; a new + /// enqueue of that id does not add another occurrence. + /// Rolling upgrades with mixed old/new processes share the same LIST. New processes + /// enqueue atomically among themselves; an old process can still race its own + /// LINSERT/RPUSH sequence and insert a duplicate until every node is upgraded. + /// Re-QueueWork while an id is still pending is a no-op (the previous multi-command + /// de-dupe intended this, but concurrent misses could both RPUSH). After dequeue, the id + /// may be queued again. + /// + /// public class RedisQueueProvider : IQueueProvider { + /// + /// Atomic unique enqueue: same LINSERT / RPUSH / LREM sequence as before, but one EVAL + /// so two concurrent misses cannot both RPUSH. LINSERT exists on Redis 2.2+; Lua on 2.6+. + /// + private const string UniqueEnqueueScript = @" +local n = redis.call('LINSERT', KEYS[1], 'BEFORE', ARGV[1], ARGV[1]) +if n == -1 or n == 0 then + redis.call('RPUSH', KEYS[1], ARGV[1]) + return 1 +end +redis.call('LREM', KEYS[1], 1, ARGV[1]) +return 0 +"; + private readonly ILogger _logger; private readonly string _connectionString; private readonly string _prefix; @@ -36,13 +69,10 @@ public async Task QueueWork(string id, QueueType queue) if (_redis == null) throw new InvalidOperationException(); - var queueName = GetQueueName(queue); - - var insertResult = await _redis.ListInsertBeforeAsync(queueName, id, id); - if (insertResult == -1 || insertResult == 0) - await _redis.ListRightPushAsync(queueName, id, When.Always); - else - await _redis.ListRemoveAsync(queueName, id, 1); + await _redis.ScriptEvaluateAsync( + UniqueEnqueueScript, + new RedisKey[] { GetQueueName(queue) }, + new RedisValue[] { id }); } public async Task DequeueWork(QueueType queue, CancellationToken cancellationToken) @@ -50,6 +80,8 @@ public async Task DequeueWork(QueueType queue, CancellationToken cancell if (_redis == null) throw new InvalidOperationException(); + // Single LPOP on the LIST. No companion membership key, so nothing can be left + // stuck (or lost) if the process dies between pop and a follow-up SREM. var result = await _redis.ListLeftPopAsync(GetQueueName(queue)); if (result.IsNull) diff --git a/test/WorkflowCore.Tests.Redis/RedisQueueProviderFixture.cs b/test/WorkflowCore.Tests.Redis/RedisQueueProviderFixture.cs new file mode 100644 index 000000000..71ae8e7d9 --- /dev/null +++ b/test/WorkflowCore.Tests.Redis/RedisQueueProviderFixture.cs @@ -0,0 +1,155 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Extensions.Logging; +using StackExchange.Redis; +using WorkflowCore.Interface; +using WorkflowCore.Providers.Redis.Services; +using Xunit; + +namespace WorkflowCore.Tests.Redis +{ + [Collection("Redis collection")] + public class RedisQueueProviderFixture : IAsyncLifetime + { + private readonly string _prefix = "q-" + Guid.NewGuid().ToString("N"); + private RedisQueueProvider _subject; + private IConnectionMultiplexer _mux; + private IDatabase _db; + + public RedisQueueProviderFixture(RedisDockerSetup dockerSetup) + { + if (dockerSetup == null) + throw new ArgumentNullException(nameof(dockerSetup)); + } + + public async Task InitializeAsync() + { + _subject = new RedisQueueProvider(RedisDockerSetup.ConnectionString, _prefix, new LoggerFactory()); + await _subject.Start(); + _mux = await ConnectionMultiplexer.ConnectAsync(RedisDockerSetup.ConnectionString); + _db = _mux.GetDatabase(); + } + + public async Task DisposeAsync() + { + if (_subject != null) + await _subject.Stop(); + if (_mux != null) + await _mux.CloseAsync(); + } + + [Fact] + public async Task QueueWork_when_id_already_pending_is_noop() + { + await _subject.QueueWork("a", QueueType.Workflow); + await _subject.QueueWork("b", QueueType.Workflow); + await _subject.QueueWork("a", QueueType.Workflow); + + (await Pending(QueueType.Workflow)).Should().Equal("a", "b"); + } + + [Fact] + public async Task QueueWork_concurrent_missing_id_leaves_one_pending_entry() + { + const string id = "same-id"; + var tasks = Enumerable.Range(0, 50) + .Select(_ => _subject.QueueWork(id, QueueType.Workflow)); + + await Task.WhenAll(tasks); + + (await Pending(QueueType.Workflow)).Should().Equal(id); + } + + [Fact] + public async Task QueueWork_does_not_add_to_preexisting_duplicate_list_entries() + { + var key = QueueKey(QueueType.Workflow); + await _db.ListRightPushAsync(key, new RedisValue[] { "dup", "dup", "dup" }); + + await _subject.QueueWork("dup", QueueType.Workflow); + + (await Pending(QueueType.Workflow)).Should().Equal("dup", "dup", "dup"); + } + + [Fact] + public async Task DequeueWork_returns_fifo_and_then_null() + { + await _subject.QueueWork("first", QueueType.Event); + await _subject.QueueWork("second", QueueType.Event); + + (await _subject.DequeueWork(QueueType.Event, CancellationToken.None)).Should().Be("first"); + (await _subject.DequeueWork(QueueType.Event, CancellationToken.None)).Should().Be("second"); + (await _subject.DequeueWork(QueueType.Event, CancellationToken.None)).Should().BeNull(); + (await Pending(QueueType.Event)).Should().BeEmpty(); + } + + [Fact] + public async Task DequeueWork_allows_id_to_be_queued_again() + { + await _subject.QueueWork("recycle", QueueType.Index); + (await _subject.DequeueWork(QueueType.Index, CancellationToken.None)).Should().Be("recycle"); + + await _subject.QueueWork("recycle", QueueType.Index); + + (await Pending(QueueType.Index)).Should().Equal("recycle"); + } + + [Fact] + public async Task DequeueWork_concurrent_pops_each_unique_id_once() + { + var ids = Enumerable.Range(0, 40).Select(i => $"id-{i}").ToArray(); + foreach (var id in ids) + await _subject.QueueWork(id, QueueType.Workflow); + + var dequeued = new ConcurrentBag(); + var workers = Enumerable.Range(0, 8).Select(async _ => + { + while (true) + { + var item = await _subject.DequeueWork(QueueType.Workflow, CancellationToken.None); + if (item == null) + return; + dequeued.Add(item); + } + }); + + await Task.WhenAll(workers); + + dequeued.Should().BeEquivalentTo(ids); + (await Pending(QueueType.Workflow)).Should().BeEmpty(); + } + + [Fact] + public async Task DequeueWork_drains_preexisting_duplicates_then_empty() + { + var key = QueueKey(QueueType.Workflow); + await _db.ListRightPushAsync(key, new RedisValue[] { "legacy", "legacy" }); + + (await _subject.DequeueWork(QueueType.Workflow, CancellationToken.None)).Should().Be("legacy"); + (await _subject.DequeueWork(QueueType.Workflow, CancellationToken.None)).Should().Be("legacy"); + (await _subject.DequeueWork(QueueType.Workflow, CancellationToken.None)).Should().BeNull(); + } + + private string QueueKey(QueueType queue) + { + switch (queue) + { + case QueueType.Workflow: return $"{_prefix}-workflows"; + case QueueType.Event: return $"{_prefix}-events"; + case QueueType.Index: return $"{_prefix}-index"; + default: throw new ArgumentOutOfRangeException(nameof(queue)); + } + } + + private async Task> Pending(QueueType queue) + { + var values = await _db.ListRangeAsync(QueueKey(queue)); + return values.Select(v => (string)v).ToList(); + } + } +}