Skip to content

Harden Redis queue uniqueness with atomic Lua enqueue #1446

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
danielgerlag wants to merge 1 commit into from
Closed

Harden Redis queue uniqueness with atomic Lua enqueue #1446

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 39 additions & 7 deletions src/providers/WorkflowCore.Providers.Redis/Services/RedisQueueProvider.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,41 @@

namespace WorkflowCore.Providers.Redis.Services
{
/// <summary>
/// Redis LIST-backed <see cref="IQueueProvider"/>. Pending ids are unique per queue.
/// </summary>
/// <remarks>
/// Compatibility:
/// <list type="bullet">
/// <item>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.</item>
/// <item>Existing LIST keys (including those that already contain duplicate entries) need no
/// migration or flush. Pre-existing duplicates drain via <see cref="DequeueWork"/>; a new
/// enqueue of that id does not add another occurrence.</item>
/// <item>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.</item>
/// <item>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.</item>
/// </list>
/// </remarks>
public class RedisQueueProvider : IQueueProvider
{
/// <summary>
/// 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+.
/// </summary>
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;
Expand All @@ -36,20 +69,19 @@ 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<string> DequeueWork(QueueType queue, CancellationToken cancellationToken)
{
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)
Expand Down
155 changes: 155 additions & 0 deletions test/WorkflowCore.Tests.Redis/RedisQueueProviderFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<string>();
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<List<string>> Pending(QueueType queue)
{
var values = await _db.ListRangeAsync(QueueKey(queue));
return values.Select(v => (string)v).ToList();
}
}
}
Loading