diff --git a/src/Foundatio.Redis/Cache/RedisCacheClient.cs b/src/Foundatio.Redis/Cache/RedisCacheClient.cs index dc4a967..ee641cc 100644 --- a/src/Foundatio.Redis/Cache/RedisCacheClient.cs +++ b/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 config) @@ -121,7 +123,7 @@ public async Task RemoveAllAsync(IEnumerable keys = null) } } } - else if (Database.Multiplexer.IsCluster()) + else if (_isCluster) { var redisKeys = keys is ICollection collection ? new List(collection.Count) : []; foreach (string key in keys.Distinct()) @@ -191,7 +193,7 @@ public async Task 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>> GetAllAsync(IEnumerable if (redisKeys.Count is 0) return ReadOnlyDictionary>.Empty; - if (_options.ConnectionMultiplexer.IsCluster()) + if (_isCluster) { var result = new Dictionary>(redisKeys.Count); // NOTE: Consider parallel processing per hash slot for performance optimization @@ -606,7 +608,7 @@ public async Task SetAllAsync(IDictionary values, TimeSpan? e pairs[index++] = new KeyValuePair(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(keyList.Count); // NOTE: Consider parallel processing per hash slot for performance optimization @@ -914,7 +916,7 @@ public async Task SetAllExpirationAsync(IDictionary 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))) diff --git a/src/Foundatio.Redis/Extensions/RedisExtensions.cs b/src/Foundatio.Redis/Extensions/RedisExtensions.cs index 0b05f19..ff35b22 100644 --- a/src/Foundatio.Redis/Extensions/RedisExtensions.cs +++ b/src/Foundatio.Redis/Extensions/RedisExtensions.cs @@ -1,4 +1,4 @@ -using System; +using System; using Foundatio.Extensions; using Foundatio.Serializer; using Foundatio.Utility; diff --git a/src/Foundatio.Redis/Messaging/RedisMessageBus.cs b/src/Foundatio.Redis/Messaging/RedisMessageBus.cs index 1793d05..8d3e4b6 100644 --- a/src/Foundatio.Redis/Messaging/RedisMessageBus.cs +++ b/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 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 + /// 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/ + /// + 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(); } diff --git a/src/Foundatio.Redis/Queues/RedisQueue.cs b/src/Foundatio.Redis/Queues/RedisQueue.cs index f6b7651..e8386ba 100644 --- a/src/Foundatio.Redis/Queues/RedisQueue.cs +++ b/src/Foundatio.Redis/Queues/RedisQueue.cs @@ -34,6 +34,7 @@ public class RedisQueue : QueueBase> 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 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) {