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(); + } + } +}