Redis queue: round-trip GroupId

README.md

@@ -66,7 +66,6 @@ await storage.SaveFileAsync("docs/readme.txt", "Hello World");
 | [AWS (S3/SQS/SNS)](https://github.com/FoundatioFx/Foundatio.AWS) | | ✅ | ✅ | ✅ | |
 | [RabbitMQ](https://github.com/FoundatioFx/Foundatio.RabbitMQ) | | | ✅ | | |
 | [Kafka](https://github.com/FoundatioFx/Foundatio.Kafka) | | | ✅ | | |
-| [Minio](https://github.com/FoundatioFx/Foundatio.Minio) | | | | ✅ | |
 | [Aliyun](https://github.com/FoundatioFx/Foundatio.Aliyun) | | | | ✅ | |
 | [SFTP](https://github.com/FoundatioFx/Foundatio.Storage.SshNet) | | | | ✅ | |
 

samples/Foundatio.SampleJobClient/Foundatio.SampleJobClient.csproj

@@ -8,7 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Foundatio.Xunit.v3" Version="13.0.4" />
+    <PackageReference Include="Foundatio.Xunit.v3" Version="13.0.5-preview.0.10" />
   </ItemGroup>
 
   <ItemGroup>

src/Foundatio.Redis/Foundatio.Redis.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <Import Project="..\..\build\common.props" />
   <ItemGroup>
-    <PackageReference Include="Foundatio" Version="13.0.4" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio" Version="13.0.5-preview.0.10" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio\Foundatio.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
 
     <PackageReference Include="StackExchange.Redis" Version="3.3.1" />

src/Foundatio.Redis/Queues/RedisQueue.cs

@@ -218,6 +218,7 @@ private string GetTopicName()
         {
             Properties = options.Properties,
             CorrelationId = options.CorrelationId,
+            GroupId = options.GroupId,
             Value = data
         };
         bool success = await _resiliencePolicy.ExecuteAsync(async _ => await _cache.AddAsync(GetPayloadKey(id), envelope, _payloadTimeToLive)).AnyContext();
@@ -240,7 +241,7 @@ await _resiliencePolicy.ExecuteAsync(async _ => await Task.WhenAll(
         }
 
         Interlocked.Increment(ref _enqueuedCount);
-        var entry = new QueueEntry<T>(id, options.CorrelationId, data, this, now, 0);
+        var entry = new QueueEntry<T>(id, options.CorrelationId, data, this, now, 0) { GroupId = options.GroupId };
         await OnEnqueuedAsync(entry).AnyContext();
 
         _logger.LogTrace("Enqueue done");
@@ -406,7 +407,7 @@ public override async Task RenewLockAsync(IQueueEntry<T> entry)
         var attemptsValue = _resiliencePolicy.ExecuteAsync(async _ => await _cache.GetAsync(GetAttemptsKey(workId), 0)).AsTask();
         await Task.WhenAll(enqueuedTimeTicks, attemptsValue).AnyContext();
 
-        var queueEntry = new QueueEntry<T>(workId, payload.Value.CorrelationId, payload.Value.Value, this, new DateTime(enqueuedTimeTicks.Result, DateTimeKind.Utc), attemptsValue.Result + 1);
+        var queueEntry = new QueueEntry<T>(workId, payload.Value.CorrelationId, payload.Value.Value, this, new DateTime(enqueuedTimeTicks.Result, DateTimeKind.Utc), attemptsValue.Result + 1) { GroupId = payload.Value.GroupId };
 
         if (payload.Value.Properties != null)
         {
@@ -894,6 +895,7 @@ private void WaitForMaintenanceTask()
 public record RedisPayloadEnvelope<T>
 {
     public string? CorrelationId { get; init; }
+    public string? GroupId { get; init; }
     public IDictionary<string, string>? Properties { get; init; }
     public required T Value { get; init; }
 }

tests/Foundatio.Redis.Tests/Extensions/ConnectionMultiplexerExtensions.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Linq;
 using System.Threading.Tasks;
 using StackExchange.Redis;
@@ -8,9 +9,17 @@ public static class ConnectionMultiplexerExtensions
 {
     public static async Task FlushAllAsync(this ConnectionMultiplexer muxer)
     {
-        int database = muxer.GetDatabase().Database;
-        foreach (var server in muxer.GetEndPoints().Select(ep => muxer.GetServer(ep)).Where(s => !s.IsReplica))
-            await server.FlushDatabaseAsync(database);
+        try
+        {
+            await FlushPrimariesAsync(muxer);
+        }
+        catch (RedisException ex) when (IsReplicaError(ex))
+        {
+            // Cluster roles can change after the multiplexer cached them (for example while the cluster settles at startup).
+            // Refresh the topology so the current primaries are flushed instead of a node that became a replica.
+            await muxer.ConfigureAsync();
+            await FlushPrimariesAsync(muxer);
+        }
     }
 
     public static async Task<long> CountAllKeysAsync(this ConnectionMultiplexer muxer)
@@ -22,4 +31,17 @@ public static async Task<long> CountAllKeysAsync(this ConnectionMultiplexer muxe
 
         return count;
     }
+
+    private static async Task FlushPrimariesAsync(ConnectionMultiplexer muxer)
+    {
+        int database = muxer.GetDatabase().Database;
+        foreach (var server in muxer.GetEndPoints().Select(ep => muxer.GetServer(ep)).Where(s => !s.IsReplica))
+            await server.FlushDatabaseAsync(database);
+    }
+
+    private static bool IsReplicaError(RedisException ex)
+    {
+        return ex.Message.Contains("replica", StringComparison.OrdinalIgnoreCase)
+            || ex.InnerException?.Message.Contains("replica", StringComparison.OrdinalIgnoreCase) is true;
+    }
 }

tests/Foundatio.Redis.Tests/Foundatio.Redis.Tests.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
     <OutputType>Exe</OutputType>
   </PropertyGroup>
@@ -8,7 +8,7 @@
     <PackageReference Include="xunit.runner.visualstudio" Version="4.0.0" />
     <PackageReference Include="GitHubActionsTestLogger" Version="3.0.5" PrivateAssets="All" />
 
-    <PackageReference Include="Foundatio.TestHarness" Version="13.0.4" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio.TestHarness" Version="13.0.5-preview.0.10" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio.TestHarness\Foundatio.TestHarness.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
   <ItemGroup>

tests/Foundatio.Redis.Tests/Queues/RedisQueueTests.cs

@@ -30,25 +30,31 @@ public RedisQueueTests(ITestOutputHelper output) : base(output)
     {
     }
 
-    protected override IQueue<SimpleWorkItem>? GetQueue(int retries = 1, TimeSpan? workItemTimeout = null, TimeSpan? retryDelay = null, int[]? retryMultipliers = null, int deadLetterMaxItems = 100, bool runQueueMaintenance = true, TimeProvider? timeProvider = null, ISerializer? serializer = null)
+    protected override IQueue<SimpleWorkItem>? GetQueue(int retries = 1, TimeSpan? workItemTimeout = null, TimeSpan? retryDelay = null, int[]? retryMultipliers = null, int deadLetterMaxItems = 100, bool runQueueMaintenance = true, TimeProvider? timeProvider = null, ISerializer? serializer = null, Func<SimpleWorkItem, string?>? groupIdResolver = null)
     {
         var muxer = SharedConnection.GetMuxer(Log);
         if (muxer is null)
             return null;
 
-        var queue = new RedisQueue<SimpleWorkItem>(o => o
-            .ConnectionMultiplexer(muxer)
-            .Retries(retries)
-            .RetryDelay(retryDelay.GetValueOrDefault(TimeSpan.FromMinutes(1)))
-            .RetryMultipliers(retryMultipliers ?? [1, 3, 5, 10])
-            .DeadLetterMaxItems(deadLetterMaxItems)
-            .TimeProvider(timeProvider ?? TimeProvider.System)
-            .WorkItemTimeout(workItemTimeout.GetValueOrDefault(TimeSpan.FromMinutes(5)))
-            .MetricsPollingInterval(TimeSpan.Zero)
-            .RunMaintenanceTasks(runQueueMaintenance)
-            .Serializer(serializer)
-            .LoggerFactory(Log)
-        );
+        var queue = new RedisQueue<SimpleWorkItem>(o =>
+        {
+            o.ConnectionMultiplexer(muxer)
+                .Retries(retries)
+                .RetryDelay(retryDelay.GetValueOrDefault(TimeSpan.FromMinutes(1)))
+                .RetryMultipliers(retryMultipliers ?? [1, 3, 5, 10])
+                .DeadLetterMaxItems(deadLetterMaxItems)
+                .TimeProvider(timeProvider ?? TimeProvider.System)
+                .WorkItemTimeout(workItemTimeout.GetValueOrDefault(TimeSpan.FromMinutes(5)))
+                .MetricsPollingInterval(TimeSpan.Zero)
+                .RunMaintenanceTasks(runQueueMaintenance)
+                .Serializer(serializer)
+                .LoggerFactory(Log);
+
+            if (groupIdResolver is not null)
+                o.GroupId(groupIdResolver);
+
+            return o;
+        });
 
         _logger.LogDebug("Queue Id: {QueueId}", queue.QueueId);
         return queue;
@@ -120,6 +126,18 @@ public override Task DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalVa
         return base.DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalValueAsync();
     }
 
+    [Fact]
+    public override Task AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync()
+    {
+        return base.AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync();
+    }
+
+    [Fact(Skip = "Redis requeues delayed retries in the minute-interval maintenance task")]
+    public override Task AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync()
+    {
+        return base.AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync();
+    }
+
     [Fact]
     public override Task DequeueAsync_WithDispose_AutoAbandonsEntryAsync()
     {
@@ -162,12 +180,84 @@ public override Task DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync(
         return base.DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync();
     }
 
+    [Fact]
+    public override Task EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync()
+    {
+        return base.EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync();
+    }
+
     [Fact]
     public override Task EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync()
     {
         return base.EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync();
     }
 
+    [Fact]
+    public override Task EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync()
+    {
+        return base.EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WhenResolverReturnsEmpty_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WhenResolverReturnsEmpty_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WhenResolverReturnsNull_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WhenResolverReturnsNull_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithActivity_AppliesCorrelationBeforeEnqueuingHandlersAsync()
+    {
+        return base.EnqueueAsync_WithActivity_AppliesCorrelationBeforeEnqueuingHandlersAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithEmptyGroupIdAndResolver_UsesPayloadValueAsync()
+    {
+        return base.EnqueueAsync_WithEmptyGroupIdAndResolver_UsesPayloadValueAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithExplicitGroupId_OverridesResolverAsync()
+    {
+        return base.EnqueueAsync_WithExplicitGroupId_OverridesResolverAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithGroupIdResolver_DoesNotChangeCallerOptionsAsync()
+    {
+        return base.EnqueueAsync_WithGroupIdResolver_DoesNotChangeCallerOptionsAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithGroupIdResolver_UsesPayloadValueAsync()
+    {
+        return base.EnqueueAsync_WithGroupIdResolver_UsesPayloadValueAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithReusedOptionsAndGroupIdResolver_EnqueuesEachMessageAsync()
+    {
+        return base.EnqueueAsync_WithReusedOptionsAndGroupIdResolver_EnqueuesEachMessageAsync();
+    }
+
     [Fact]
     public override Task EnqueueAsync_WithUniqueId_UsesProvidedIdAsync()
     {