Use sharded pub/sub for Redis Cluster to fix duplicate message delivery

src/Foundatio.Redis/Cache/RedisCacheClient.cs

@@ -21,6 +21,7 @@ public sealed class RedisCacheClient : ICacheClient, IHaveSerializer
     private readonly RedisCacheClientOptions _options;
     private readonly TimeProvider _timeProvider;
     private readonly ILogger _logger;
+    private readonly bool _isCluster;
 
     private readonly AsyncLock _lock = new();
     private bool _scriptsLoaded;
@@ -43,6 +44,7 @@ public RedisCacheClient(RedisCacheClientOptions options)
 
         options.ConnectionMultiplexer.ConnectionRestored += ConnectionMultiplexerOnConnectionRestored;
         options.ConnectionMultiplexer.ConnectionFailed += ConnectionMultiplexerOnConnectionFailed;
+        _isCluster = options.ConnectionMultiplexer.IsCluster();
     }
 
     public RedisCacheClient(Builder<RedisCacheClientOptionsBuilder, RedisCacheClientOptions> config)
@@ -121,7 +123,7 @@ public async Task<int> RemoveAllAsync(IEnumerable<string> keys = null)
                 }
             }
         }
-        else if (Database.Multiplexer.IsCluster())
+        else if (_isCluster)
         {
             var redisKeys = keys is ICollection<string> collection ? new List<RedisKey>(collection.Count) : [];
             foreach (string key in keys.Distinct())
@@ -191,7 +193,7 @@ public async Task<int> RemoveByPrefixAsync(string prefix)
             return 0;
 
         const int batchSize = 250;
-        bool isCluster = _options.ConnectionMultiplexer.IsCluster();
+        bool isCluster = _isCluster;
 
         long deleted = 0;
         foreach (var endpoint in endpoints)
@@ -298,7 +300,7 @@ public async Task<IDictionary<string, CacheValue<T>>> GetAllAsync<T>(IEnumerable
         if (redisKeys.Count is 0)
             return ReadOnlyDictionary<string, CacheValue<T>>.Empty;
 
-        if (_options.ConnectionMultiplexer.IsCluster())
+        if (_isCluster)
         {
             var result = new Dictionary<string, CacheValue<T>>(redisKeys.Count);
             // NOTE: Consider parallel processing per hash slot for performance optimization
@@ -606,7 +608,7 @@ public async Task<int> SetAllAsync<T>(IDictionary<string, T> values, TimeSpan? e
             pairs[index++] = new KeyValuePair<RedisKey, RedisValue>(kvp.Key, kvp.Value.ToRedisValue(_options.Serializer));
         }
 
-        if (_options.ConnectionMultiplexer.IsCluster())
+        if (_isCluster)
         {
             // For cluster/sentinel, group keys by hash slot since batch operations
             // require all keys to be in the same slot
@@ -834,7 +836,7 @@ public Task SetExpirationAsync(string key, TimeSpan expiresIn)
 
         await LoadScriptsAsync().AnyContext();
 
-        if (_options.ConnectionMultiplexer.IsCluster())
+        if (_isCluster)
         {
             var result = new Dictionary<string, TimeSpan?>(keyList.Count);
             // NOTE: Consider parallel processing per hash slot for performance optimization
@@ -914,7 +916,7 @@ public async Task SetAllExpirationAsync(IDictionary<string, TimeSpan?> expiratio
 
         await LoadScriptsAsync().AnyContext();
 
-        if (_options.ConnectionMultiplexer.IsCluster())
+        if (_isCluster)
         {
             // NOTE: Consider parallel processing per hash slot for performance optimization
             foreach (var hashSlotGroup in expirations.GroupBy(kvp => _options.ConnectionMultiplexer.HashSlot(kvp.Key)))

src/Foundatio.Redis/Extensions/RedisExtensions.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using Foundatio.Extensions;
 using Foundatio.Serializer;
 using Foundatio.Utility;

src/Foundatio.Redis/Messaging/RedisMessageBus.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 using Foundatio.AsyncEx;
 using Foundatio.Extensions;
+using Foundatio.Redis;
 using Foundatio.Serializer;
 using Microsoft.Extensions.Logging;
 using StackExchange.Redis;
@@ -16,6 +17,7 @@ public class RedisMessageBus : MessageBusBase<RedisMessageBusOptions>
     private readonly AsyncLock _lock = new();
     private bool _isSubscribed;
     private ChannelMessageQueue _channelMessageQueue;
+    private RedisChannel? _channel;
 
     public RedisMessageBus(RedisMessageBusOptions options) : base(options)
     {
@@ -26,6 +28,24 @@ public RedisMessageBus(Builder<RedisMessageBusOptionsBuilder, RedisMessageBusOpt
     {
     }
 
+    /// <summary>
+    /// Gets the Redis channel for pub/sub. In Redis Cluster mode, uses sharded pub/sub
+    /// (SPUBLISH/SSUBSCRIBE) to avoid duplicate message delivery caused by cluster-wide
+    /// broadcast. Regular PUBLISH in a cluster broadcasts to all nodes, and StackExchange.Redis
+    /// spreads Literal subscriptions across nodes, which can cause subscribers to receive
+    /// the same message multiple times from different primaries. Sharded pub/sub routes all
+    /// operations for a given channel through a single shard, preventing per-primary duplicate
+    /// delivery while preserving full fanout to all subscribers on that shard. This does not
+    /// provide end-to-end exactly-once semantics; callers must remain tolerant of lost or
+    /// duplicated messages across disconnects and retries.
+    /// Falls back to standard PUBLISH/SUBSCRIBE for standalone, sentinel, and proxy deployments.
+    /// See: https://redis.io/docs/latest/commands/spublish/
+    /// See: https://redis.io/docs/latest/commands/ssubscribe/
+    /// </summary>
+    private RedisChannel Channel => _channel ??= _options.Subscriber.Multiplexer.IsCluster()
+        ? RedisChannel.Sharded(_options.Topic)
+        : RedisChannel.Literal(_options.Topic);
+
     protected override async Task EnsureTopicSubscriptionAsync(CancellationToken cancellationToken)
     {
         if (_isSubscribed)
@@ -37,7 +57,7 @@ protected override async Task EnsureTopicSubscriptionAsync(CancellationToken can
                 return;
 
             _logger.LogTrace("Subscribing to topic: {Topic}", _options.Topic);
-            _channelMessageQueue = await _options.Subscriber.SubscribeAsync(RedisChannel.Literal(_options.Topic)).AnyContext();
+            _channelMessageQueue = await _options.Subscriber.SubscribeAsync(Channel).AnyContext();
             _channelMessageQueue.OnMessage(OnMessage);
             _isSubscribed = true;
             _logger.LogTrace("Subscribed to topic: {Topic}", _options.Topic);
@@ -117,7 +137,7 @@ protected override async Task PublishImplAsync(string messageType, object messag
         // TODO: Use ILockProvider to lock on UniqueId to ensure it doesn't get duplicated
         // Wrap only the transport call in resilience policy
         await _resiliencePolicy.ExecuteAsync(async _ =>
-            await _options.Subscriber.PublishAsync(RedisChannel.Literal(_options.Topic), data, CommandFlags.FireAndForget),
+            await _options.Subscriber.PublishAsync(Channel, data, CommandFlags.FireAndForget),
             cancellationToken).AnyContext();
     }
 

src/Foundatio.Redis/Queues/RedisQueue.cs

@@ -34,6 +34,7 @@ public class RedisQueue<T> : QueueBase<T, RedisQueueOptions<T>> where T : class
     private readonly TimeSpan _payloadTimeToLive;
     private bool _scriptsLoaded;
     private readonly string _listPrefix;
+    private readonly RedisChannel _topicChannel;
 
     private LoadedLuaScript _dequeueId;
 
@@ -49,11 +50,15 @@ public RedisQueue(RedisQueueOptions<T> options) : base(options)
         _payloadTimeToLive = GetPayloadTtl();
         _subscriber = _options.ConnectionMultiplexer.GetSubscriber();
 
-        _listPrefix = _options.ConnectionMultiplexer.IsCluster() ? "{q:" + _options.Name + "}" : $"q:{_options.Name}";
+        bool isCluster = _options.ConnectionMultiplexer.IsCluster();
+        _listPrefix = isCluster ? "{q:" + _options.Name + "}" : $"q:{_options.Name}";
         _queueListName = $"{_listPrefix}:in";
         _workListName = $"{_listPrefix}:work";
         _waitListName = $"{_listPrefix}:wait";
         _deadListName = $"{_listPrefix}:dead";
+        _topicChannel = isCluster
+            ? RedisChannel.Sharded(_queueListName)
+            : RedisChannel.Literal(_queueListName);
 
         // min is 1 second, max is 1 minute
         var interval = _options.WorkItemTimeout > TimeSpan.FromSeconds(1) ? _options.WorkItemTimeout.Min(TimeSpan.FromMinutes(1)) : TimeSpan.FromSeconds(1);
@@ -98,7 +103,7 @@ private async Task EnsureTopicSubscriptionAsync()
                 return;
 
             _logger.LogTrace("Subscribing to enqueue messages for {QueueName}", _options.Name);
-            await _subscriber.SubscribeAsync(RedisChannel.Literal(GetTopicName()), OnTopicMessage).AnyContext();
+            await _subscriber.SubscribeAsync(_topicChannel, OnTopicMessage).AnyContext();
             _isSubscribed = true;
             _logger.LogTrace("Subscribed to enqueue messages for {QueueName}", _options.Name);
         }
@@ -224,7 +229,7 @@ await _resiliencePolicy.ExecuteAsync(async _ => await Task.WhenAll(
         try
         {
             _autoResetEvent.Set();
-            await _resiliencePolicy.ExecuteAsync(async _ => await _subscriber.PublishAsync(RedisChannel.Literal(GetTopicName()), id)).AnyContext();
+            await _resiliencePolicy.ExecuteAsync(async _ => await _subscriber.PublishAsync(_topicChannel, id)).AnyContext();
         }
         catch (Exception ex)
         {
@@ -524,7 +529,7 @@ await _resiliencePolicy.ExecuteAsync(async _ => await Task.WhenAll(
             await _resiliencePolicy.ExecuteAsync(async _ => await Task.WhenAll(
                 Database.KeyDeleteAsync(GetDequeuedTimeKey(entry.Id)),
                 // This should pulse the monitor.
-                _subscriber.PublishAsync(RedisChannel.Literal(GetTopicName()), entry.Id)
+                _subscriber.PublishAsync(_topicChannel, entry.Id)
             )).AnyContext();
         }
 
@@ -704,7 +709,7 @@ public async Task DoMaintenanceWorkAsync()
                 if (!success)
                     throw new Exception("Unable to move item to queue list.");
 
-                await _resiliencePolicy.ExecuteAsync(async _ => await _subscriber.PublishAsync(RedisChannel.Literal(GetTopicName()), waitId), cancellationToken: DisposedCancellationToken).AnyContext();
+                await _resiliencePolicy.ExecuteAsync(async _ => await _subscriber.PublishAsync(_topicChannel, waitId), cancellationToken: DisposedCancellationToken).AnyContext();
             }
         }
         catch (Exception ex)
@@ -781,7 +786,7 @@ public override void Dispose()
                     _logger.LogTrace("Unsubscribing from topic {Topic}", GetTopicName());
                     try
                     {
-                        _subscriber.Unsubscribe(RedisChannel.Literal(GetTopicName()), OnTopicMessage, CommandFlags.FireAndForget);
+                        _subscriber.Unsubscribe(_topicChannel, OnTopicMessage, CommandFlags.FireAndForget);
                     }
                     catch (Exception ex)
                     {