From 44816153820ed514d2a6e22f3da24c32747b8b50 Mon Sep 17 00:00:00 2001 From: richard-churchman Date: Sun, 22 Mar 2026 11:02:43 +0100 Subject: [PATCH] Implemented Redis Resiliency and improved failover capabilities: * Replaced the StackOverflow Redis client in the Cache library with the Resilient Redis Library. * Improved policy wrappers by introducing two distinct policies: one for idempotent operations and one for increment operations. Added a circuit breaker to the Redis Resiliency layer to prevent thundering herd on re-instantiation, along with shared state tracking for known-down Redis instances. * Updated the Redis resilient command wrapper to proxy all commands through the appropriate policy (e.g. based on idempotency). * Implemented certain other Redis policy state detection. --- Jube.Cache/CacheService.cs | 5 +- .../Redis/CacheAbstractionRepository.cs | 3 +- .../Redis/CacheCallbackPublishSubscribe.cs | 5 +- .../Redis/CachePayloadLatestRepository.cs | 9 +- Jube.Cache/Redis/CachePayloadRepository.cs | 13 +- Jube.Cache/Redis/CacheReferenceDate.cs | 3 +- Jube.Cache/Redis/CacheSanctionRepository.cs | 3 +- .../Redis/CacheTtlCounterEntryRepository.cs | 5 +- Jube.Cache/Redis/CacheTtlCounterRepository.cs | 3 +- Jube.Cache/Redis/CacheWalRepository.cs | 3 +- .../ResilientRedisConnection.cs | 66 ++++-- .../ResilientRedisDatabase.cs | 219 +++++++++++------- 12 files changed, 218 insertions(+), 119 deletions(-) diff --git a/Jube.Cache/CacheService.cs b/Jube.Cache/CacheService.cs index 923e4b6..61c43e1 100644 --- a/Jube.Cache/CacheService.cs +++ b/Jube.Cache/CacheService.cs @@ -17,6 +17,7 @@ namespace Jube.Cache using log4net; using Redis; using Redis.Callback; + using ResilientRedisConnection; using StackExchange.Redis; using TaskCancellation; @@ -52,12 +53,12 @@ namespace Jube.Cache ConnectionMultiplexer = ConnectionMultiplexer.Connect(redisConnectionString); - RedisDatabase = ConnectionMultiplexer.GetDatabase(); + RedisDatabase = new ResilientRedisConnection(ConnectionMultiplexer, log).GetDatabase(); } public Task InstantiateRepositoriesTask { get; set; } public ConnectionMultiplexer ConnectionMultiplexer { get; set; } - public IDatabase RedisDatabase { get; set; } + public ResilientRedisDatabase RedisDatabase { get; set; } public CacheAbstractionRepository CacheAbstractionRepository { get; set; } public CachePayloadLatestRepository CachePayloadLatestRepository { get; set; } public CachePayloadRepository CachePayloadRepository { get; set; } diff --git a/Jube.Cache/Redis/CacheAbstractionRepository.cs b/Jube.Cache/Redis/CacheAbstractionRepository.cs index a092a12..ccbe874 100644 --- a/Jube.Cache/Redis/CacheAbstractionRepository.cs +++ b/Jube.Cache/Redis/CacheAbstractionRepository.cs @@ -16,10 +16,11 @@ namespace Jube.Cache.Redis using Interfaces; using log4net; using Models; + using ResilientRedisConnection; using StackExchange.Redis; public class CacheAbstractionRepository( - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, ILog log, CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheAbstractionRepository { diff --git a/Jube.Cache/Redis/CacheCallbackPublishSubscribe.cs b/Jube.Cache/Redis/CacheCallbackPublishSubscribe.cs index 3329dc6..9a4b77b 100644 --- a/Jube.Cache/Redis/CacheCallbackPublishSubscribe.cs +++ b/Jube.Cache/Redis/CacheCallbackPublishSubscribe.cs @@ -16,6 +16,7 @@ namespace Jube.Cache.Redis using System.Collections.Concurrent; using System.Net; using log4net; + using ResilientRedisConnection; using StackExchange.Redis; using TaskCancellation; @@ -30,10 +31,10 @@ namespace Jube.Cache.Redis private readonly ConnectionMultiplexer connectionMultiplexer; private readonly string localCacheInstanceGuidString; private readonly ILog log; - private readonly IDatabaseAsync redisDatabase; + private readonly ResilientRedisDatabase redisDatabase; public CacheCallbackPublishSubscribe(ConnectionMultiplexer connectionMultiplexer, - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, ConcurrentDictionary> callbacks, int callbackTimeout, ILog log, diff --git a/Jube.Cache/Redis/CachePayloadLatestRepository.cs b/Jube.Cache/Redis/CachePayloadLatestRepository.cs index 2da33a9..9ea9118 100644 --- a/Jube.Cache/Redis/CachePayloadLatestRepository.cs +++ b/Jube.Cache/Redis/CachePayloadLatestRepository.cs @@ -25,13 +25,14 @@ namespace Jube.Cache.Redis using log4net; using MessagePack; using Models; + using ResilientRedisConnection; using Serialization; using StackExchange.Redis; using TaskCancellation.TaskHelper; public class CachePayloadLatestRepository( string postgresConnectionString, - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, ILog log, CommandFlags commandFlag = CommandFlags.FireAndForget) : ICachePayloadLatestRepository { @@ -191,7 +192,7 @@ namespace Jube.Cache.Redis var updateLatestTask = redisDatabase.SortedSetUpdateAsync(redisKeyReferenceDateLatest, redisHSetKey, referenceDateTimestamp); - await redisDatabase.HashIncrementAsync(redisKeyPayloadLatestCount, entryKey); + await redisDatabase.HashIncrementAsync(redisKeyPayloadLatestCount, entryKey, 1); await redisDatabase.HashSetAsync(redisKeyPayloadLatest, redisHSetKey, bytes); await updateLatestTask; @@ -286,7 +287,7 @@ namespace Jube.Cache.Redis } } - private static async Task BatchSortedSetRemoveAsync(IDatabaseAsync db, RedisKey key, IEnumerable values, CommandFlags flags) + private static async Task BatchSortedSetRemoveAsync(ResilientRedisDatabase db, RedisKey key, IEnumerable values, CommandFlags flags) { const int batchSize = 1000; var valuesArray = values.ToArray(); @@ -300,7 +301,7 @@ namespace Jube.Cache.Redis } } - private static async Task BatchHashDeleteAsync(IDatabaseAsync db, RedisKey key, IEnumerable fields, CommandFlags flags) + private static async Task BatchHashDeleteAsync(ResilientRedisDatabase db, RedisKey key, IEnumerable fields, CommandFlags flags) { const int batchSize = 1000; var fieldsArray = fields.ToArray(); diff --git a/Jube.Cache/Redis/CachePayloadRepository.cs b/Jube.Cache/Redis/CachePayloadRepository.cs index 397d181..090b12b 100644 --- a/Jube.Cache/Redis/CachePayloadRepository.cs +++ b/Jube.Cache/Redis/CachePayloadRepository.cs @@ -24,6 +24,7 @@ namespace Jube.Cache.Redis using Interfaces; using log4net; using MessagePack; + using ResilientRedisConnection; using Serialization; using Serialization.DictionaryNoBoxing.MessagePack; using StackExchange.Redis; @@ -43,7 +44,7 @@ namespace Jube.Cache.Redis private readonly MessagePackSerializerOptions messagePackSerializerOptions; private readonly string postgresConnectionString; private readonly bool publishSubscribe; - private readonly IDatabaseAsync redisDatabase; + private readonly ResilientRedisDatabase redisDatabase; private readonly bool storePayloadCountsAndBytes; private readonly SemaphoreSlim timerSemaphore = new SemaphoreSlim(1, 1); private LocalCacheInstance localCacheInstance; @@ -52,7 +53,7 @@ namespace Jube.Cache.Redis private LruCacheConcurrentSizedDictionary lruCacheConcurrentSizedDictionary; private Timer timer; - private CachePayloadRepository(ConnectionMultiplexer connectionMultiplexer, IDatabaseAsync redisDatabase, + private CachePayloadRepository(ConnectionMultiplexer connectionMultiplexer, ResilientRedisDatabase redisDatabase, string postgresConnectionString, ILog log, CommandFlags commandFlag, bool fill, bool localCache, long localCacheBytes, bool messagePackCompression, bool storePayloadCountsAndBytes, bool publishSubscribe, CancellationToken token = default) @@ -120,7 +121,7 @@ namespace Jube.Cache.Redis { var redisKeyPayloadCount = $"PayloadCount:{tenantRegistryId}"; var redisKeyPayloadBytes = $"PayloadBytes:{tenantRegistryId}"; - tasks.Add(redisDatabase.HashIncrementAsync(redisKeyPayloadCount, entityAnalysisModelGuid.ToString("N"))); + tasks.Add(redisDatabase.HashIncrementAsync(redisKeyPayloadCount, entityAnalysisModelGuid.ToString("N"), 1)); tasks.Add(redisDatabase.HashIncrementAsync(redisKeyPayloadBytes, entityAnalysisModelGuid.ToString("N"), bytes.Length)); } @@ -299,7 +300,7 @@ namespace Jube.Cache.Redis public static async Task CreateAsync( ConnectionMultiplexer connectionMultiplexer, - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, string postgresConnectionString, ILog log, CommandFlags commandFlag, @@ -826,7 +827,7 @@ namespace Jube.Cache.Redis ))); tasks.Add(TaskHelper.MeasureTaskTimeAndMemoryAllocatedAsync(TaskType.HashDecrementCount, async () => await redisDatabase.HashDecrementAsync( - redisKeyCount, redisHashKeyForRedisKey + redisKeyCount, redisHashKeyForRedisKey, 1 ))); tasks.Add(TaskHelper.MeasureTaskTimeAndMemoryAllocatedAsync(TaskType.HashDeletePayload, async () => await redisDatabase.HashDeleteAsync( @@ -1073,7 +1074,7 @@ namespace Jube.Cache.Redis } - private async Task BatchSortedSetRemoveAsync(IDatabaseAsync db, RedisKey key, IEnumerable values) + private async Task BatchSortedSetRemoveAsync(ResilientRedisDatabase db, RedisKey key, IEnumerable values) { const int batchSize = 1000; var valuesArray = values.ToArray(); diff --git a/Jube.Cache/Redis/CacheReferenceDate.cs b/Jube.Cache/Redis/CacheReferenceDate.cs index dc6699e..6bf18a5 100644 --- a/Jube.Cache/Redis/CacheReferenceDate.cs +++ b/Jube.Cache/Redis/CacheReferenceDate.cs @@ -16,10 +16,11 @@ namespace Jube.Cache.Redis using Extensions; using Interfaces; using log4net; + using ResilientRedisConnection; using StackExchange.Redis; public class CacheReferenceDate( - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, ILog log, CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheReferenceDate { diff --git a/Jube.Cache/Redis/CacheSanctionRepository.cs b/Jube.Cache/Redis/CacheSanctionRepository.cs index a7ca288..0f3d077 100644 --- a/Jube.Cache/Redis/CacheSanctionRepository.cs +++ b/Jube.Cache/Redis/CacheSanctionRepository.cs @@ -17,11 +17,12 @@ namespace Jube.Cache.Redis using log4net; using MessagePack; using Models; + using ResilientRedisConnection; using Serialization; using StackExchange.Redis; public class CacheSanctionRepository( - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, ILog log, CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheSanctionRepository { diff --git a/Jube.Cache/Redis/CacheTtlCounterEntryRepository.cs b/Jube.Cache/Redis/CacheTtlCounterEntryRepository.cs index bed17ad..82df513 100644 --- a/Jube.Cache/Redis/CacheTtlCounterEntryRepository.cs +++ b/Jube.Cache/Redis/CacheTtlCounterEntryRepository.cs @@ -17,10 +17,11 @@ namespace Jube.Cache.Redis using Interfaces; using log4net; using Models; + using ResilientRedisConnection; using StackExchange.Redis; public class CacheTtlCounterEntryRepository( - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, ILog log, CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheTtlCounterEntryRepository { @@ -47,7 +48,7 @@ namespace Jube.Cache.Redis { continue; } - + if (keyTtlCounterEntry.Value.HasValue) { expired.Add(new ExpiredTtlCounterEntry diff --git a/Jube.Cache/Redis/CacheTtlCounterRepository.cs b/Jube.Cache/Redis/CacheTtlCounterRepository.cs index bb60124..125036b 100644 --- a/Jube.Cache/Redis/CacheTtlCounterRepository.cs +++ b/Jube.Cache/Redis/CacheTtlCounterRepository.cs @@ -15,10 +15,11 @@ namespace Jube.Cache.Redis { using Interfaces; using log4net; + using ResilientRedisConnection; using StackExchange.Redis; public class CacheTtlCounterRepository( - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, ILog log, CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheTtlCounterRepository { diff --git a/Jube.Cache/Redis/CacheWalRepository.cs b/Jube.Cache/Redis/CacheWalRepository.cs index ccc09a6..54a57df 100644 --- a/Jube.Cache/Redis/CacheWalRepository.cs +++ b/Jube.Cache/Redis/CacheWalRepository.cs @@ -15,10 +15,11 @@ namespace Jube.Cache.Redis { using Interfaces; using log4net; + using ResilientRedisConnection; using StackExchange.Redis; public class CacheWalRepository( - IDatabaseAsync redisDatabase, + ResilientRedisDatabase redisDatabase, ILog log, CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheWalRepository { diff --git a/Jube.ResilientRedisConnection/ResilientRedisConnection.cs b/Jube.ResilientRedisConnection/ResilientRedisConnection.cs index 1c96ac4..8ec43da 100644 --- a/Jube.ResilientRedisConnection/ResilientRedisConnection.cs +++ b/Jube.ResilientRedisConnection/ResilientRedisConnection.cs @@ -21,36 +21,66 @@ namespace Jube.ResilientRedisConnection { private readonly IConnectionMultiplexer multiplexer; - public ResilientRedisConnection(IConnectionMultiplexer multiplexer, ILog log, int maxRetries = 10) + public ResilientRedisConnection(IConnectionMultiplexer multiplexer, ILog log, int maxRetries = 5) { this.multiplexer = multiplexer ?? throw new ArgumentNullException(nameof(multiplexer)); - FailoverPolicy = Policy + var circuitBreaker = Policy .Handle() .Or() .Or(ex => - ex.Message.StartsWith("LOADING", StringComparison.OrdinalIgnoreCase) || - ex.Message.StartsWith("MASTERDOWN", StringComparison.OrdinalIgnoreCase) || - ex.Message.StartsWith("CLUSTERDOWN", StringComparison.OrdinalIgnoreCase) || - ex.Message.StartsWith("TRYAGAIN", StringComparison.OrdinalIgnoreCase) || - ex.Message.StartsWith("MOVED", StringComparison.OrdinalIgnoreCase) || - ex.Message.StartsWith("ASK", StringComparison.OrdinalIgnoreCase)) + ex.Message.Contains("LOADING", StringComparison.OrdinalIgnoreCase) || + ex.Message.Contains("MASTERDOWN", StringComparison.OrdinalIgnoreCase)) + .CircuitBreakerAsync(5, TimeSpan.FromSeconds(30)); + + var fullRetry = Policy + .Handle() + .Or() + .Or(ex => + ex.Message.Contains("LOADING", StringComparison.OrdinalIgnoreCase) || + ex.Message.Contains("MASTERDOWN", StringComparison.OrdinalIgnoreCase) || + ex.Message.Contains("CLUSTERDOWN", StringComparison.OrdinalIgnoreCase) || + ex.Message.Contains("TRYAGAIN", StringComparison.OrdinalIgnoreCase)) .Or(ex => ex.Message.Contains("No connection is available", StringComparison.OrdinalIgnoreCase) || - ex.Message.Contains("multiplexer is not connected", StringComparison.OrdinalIgnoreCase) || - ex.Message.Contains("The message was already in a completed state", StringComparison.OrdinalIgnoreCase)) + ex.Message.Contains("multiplexer is not connected", StringComparison.OrdinalIgnoreCase)) .WaitAndRetryAsync( maxRetries, - attempt => - TimeSpan.FromSeconds(Math.Min(Math.Pow(2, attempt), 10)) + - TimeSpan.FromMilliseconds(Random.Shared.Next(0, 500)), - (ex, delay, retryCount, _) => + attempt => TimeSpan.FromSeconds(Math.Min(Math.Pow(2, attempt), 10)) + + TimeSpan.FromMilliseconds(Random.Shared.Next(0, 500)), + (ex, _, count, _) => { - log.Warn($"Redis failover: attempt {retryCount} of {maxRetries}, " + - $"waiting {delay.TotalSeconds:F1}s. {ex.Message}"); + if (count == maxRetries) + { + log.Warn($"Redis Retry threshold hit: {count}/{maxRetries}. {ex.Message}"); + } }); + + var connectionRetry = Policy + .Handle() + .Or(ex => + ex.Message.Contains("READONLY", StringComparison.OrdinalIgnoreCase) || + ex.Message.Contains("TRYAGAIN", StringComparison.OrdinalIgnoreCase)) + .Or(ex => + ex.Message.Contains("No connection is available", StringComparison.OrdinalIgnoreCase)) + .WaitAndRetryAsync( + maxRetries, + attempt => TimeSpan.FromSeconds(Math.Min(Math.Pow(2, attempt), 10)) + + TimeSpan.FromMilliseconds(Random.Shared.Next(0, 500)), + (ex, _, count, _) => + { + if (count == maxRetries) + { + log.Warn($"Redis Retry threshold hit: {count}/{maxRetries}. {ex.Message}"); + } + } + ); + + IdempotentPolicy = Policy.WrapAsync(circuitBreaker, fullRetry); + NonIdempotentPolicy = Policy.WrapAsync(circuitBreaker, connectionRetry); } - private IAsyncPolicy FailoverPolicy { get; } + private IAsyncPolicy IdempotentPolicy { get; } + private IAsyncPolicy NonIdempotentPolicy { get; } public void Dispose() { @@ -59,7 +89,7 @@ namespace Jube.ResilientRedisConnection public ResilientRedisDatabase GetDatabase(int db = -1) { - return new ResilientRedisDatabase(multiplexer.GetDatabase(db), FailoverPolicy); + return new ResilientRedisDatabase(multiplexer.GetDatabase(db), IdempotentPolicy, NonIdempotentPolicy); } } } diff --git a/Jube.ResilientRedisConnection/ResilientRedisDatabase.cs b/Jube.ResilientRedisConnection/ResilientRedisDatabase.cs index 1743ce7..9f023ed 100644 --- a/Jube.ResilientRedisConnection/ResilientRedisDatabase.cs +++ b/Jube.ResilientRedisConnection/ResilientRedisDatabase.cs @@ -16,163 +16,222 @@ namespace Jube.ResilientRedisConnection using Polly; using StackExchange.Redis; - public class ResilientRedisDatabase(IDatabase inner, IAsyncPolicy policy) + public class ResilientRedisDatabase(IDatabase inner, IAsyncPolicy idempotentPolicy, IAsyncPolicy nonIdempotentPolicy) { - private readonly IAsyncPolicy policy = policy ?? throw new ArgumentNullException(nameof(policy)); + private readonly IAsyncPolicy idempotentPolicy = idempotentPolicy ?? throw new ArgumentNullException(nameof(idempotentPolicy)); - public IDatabase UnderlyingDatabase + private IDatabase UnderlyingDatabase { get; } = inner ?? throw new ArgumentNullException(nameof(inner)); - public Task StringSetAsync(RedisKey key, RedisValue value, - TimeSpan? expiry = null, When when = When.Always, CommandFlags flags = CommandFlags.None) - { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.StringSetAsync(key, value, expiry, when, flags), - new Context()); - } - - public Task StringGetAsync(RedisKey key, CommandFlags flags = CommandFlags.None) - { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.StringGetAsync(key, flags), - new Context()); - } - - public Task StringSetAsync(KeyValuePair[] values, - When when = When.Always, CommandFlags flags = CommandFlags.None) - { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.StringSetAsync(values, when, flags), - new Context()); - } - - public Task StringGetAsync(RedisKey[] keys, CommandFlags flags = CommandFlags.None) - { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.StringGetAsync(keys, flags), - new Context()); - } - - public Task KeyDeleteAsync(RedisKey key, CommandFlags flags = CommandFlags.None) - { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.KeyDeleteAsync(key, flags), - new Context()); - } - public Task KeyExistsAsync(RedisKey key, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => + return idempotentPolicy.ExecuteAsync(_ => UnderlyingDatabase.KeyExistsAsync(key, flags), new Context()); } - public Task KeyExpireAsync(RedisKey key, TimeSpan? expiry, CommandFlags flags = CommandFlags.None) + private Task KeyRenameAsync(RedisKey key, RedisKey newKey, When when = When.Always, + CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.KeyExpireAsync(key, expiry, flags), + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.KeyRenameAsync(key, newKey, when, flags), new Context()); } + public bool KeyRename(RedisKey key, RedisKey newKey, When when = When.Always, + CommandFlags flags = CommandFlags.None) + { + return KeyRenameAsync(key, newKey, when, flags).GetAwaiter().GetResult(); + } + public Task HashSetAsync(RedisKey key, RedisValue field, RedisValue value, When when = When.Always, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => + return idempotentPolicy.ExecuteAsync(_ => UnderlyingDatabase.HashSetAsync(key, field, value, when, flags), new Context()); } - public Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None) + private Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => + return idempotentPolicy.ExecuteAsync(_ => UnderlyingDatabase.HashSetAsync(key, hashFields, flags), new Context()); } + public void HashSet(RedisKey key, RedisValue field, RedisValue value, + When when = When.Always, CommandFlags flags = CommandFlags.None) + { + HashSetAsync(key, field, value, when, flags).GetAwaiter().GetResult(); + } + + public void HashSet(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None) + { + HashSetAsync(key, hashFields, flags).GetAwaiter().GetResult(); + } + public Task HashGetAsync(RedisKey key, RedisValue field, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => + return idempotentPolicy.ExecuteAsync(_ => UnderlyingDatabase.HashGetAsync(key, field, flags), new Context()); } - public Task HashGetAllAsync(RedisKey key, CommandFlags flags = CommandFlags.None) + public Task HashGetAsync(RedisKey key, RedisValue[] fields, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.HashGetAllAsync(key, flags), + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.HashGetAsync(key, fields, flags), new Context()); } public Task HashDeleteAsync(RedisKey key, RedisValue field, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => + return idempotentPolicy.ExecuteAsync(_ => UnderlyingDatabase.HashDeleteAsync(key, field, flags), new Context()); } - public Task ListRightPushAsync(RedisKey key, RedisValue value, When when = When.Always, + public Task HashDeleteAsync(RedisKey key, RedisValue[] fields, CommandFlags flags = CommandFlags.None) + { + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.HashDeleteAsync(key, fields, flags), + new Context()); + } + + public bool HashDelete(RedisKey key, RedisValue field, CommandFlags flags = CommandFlags.None) + { + return HashDeleteAsync(key, field, flags).GetAwaiter().GetResult(); + } + + public Task HashIncrementAsync(RedisKey key, RedisValue field, long value, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.ListRightPushAsync(key, value, when, flags), + return nonIdempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.HashIncrementAsync(key, field, value, flags), new Context()); } - public Task ListLeftPopAsync(RedisKey key, CommandFlags flags = CommandFlags.None) + public Task HashIncrementAsync(RedisKey key, RedisValue field, double value, + CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.ListLeftPopAsync(key, flags), + return nonIdempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.HashIncrementAsync(key, field, value, flags), new Context()); } - public Task ListLengthAsync(RedisKey key, CommandFlags flags = CommandFlags.None) + public Task HashDecrementAsync(RedisKey key, RedisValue field, long value, + CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.ListLengthAsync(key, flags), + return nonIdempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.HashDecrementAsync(key, field, value, flags), + new Context()); + } + + public Task HashDecrementAsync(RedisKey key, RedisValue field, double value, + CommandFlags flags = CommandFlags.None) + { + return nonIdempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.HashDecrementAsync(key, field, value, flags), + new Context()); + } + + public Task HashStringLengthAsync(RedisKey key, RedisValue field, CommandFlags flags = CommandFlags.None) + { + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.HashStringLengthAsync(key, field, flags), + new Context()); + } + + public IAsyncEnumerable HashScanAsync(RedisKey key, RedisValue pattern = default, + int pageSize = 250, long cursor = 0, int pageOffset = 0, CommandFlags flags = CommandFlags.None) + { + return UnderlyingDatabase.HashScanAsync(key, pattern, pageSize, cursor, pageOffset, flags); + } + + public IEnumerable HashScan(RedisKey key, RedisValue pattern = default, + int pageSize = 250, long cursor = 0, int pageOffset = 0, CommandFlags flags = CommandFlags.None) + { + return UnderlyingDatabase.HashScan(key, pattern, pageSize, cursor, pageOffset, flags); + } + + public Task SetAddAsync(RedisKey key, RedisValue value, CommandFlags flags = CommandFlags.None) + { + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.SetAddAsync(key, value, flags), + new Context()); + } + + public Task SetRemoveAsync(RedisKey key, RedisValue value, CommandFlags flags = CommandFlags.None) + { + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.SetRemoveAsync(key, value, flags), + new Context()); + } + + public Task SetMembersAsync(RedisKey key, CommandFlags flags = CommandFlags.None) + { + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.SetMembersAsync(key, flags), new Context()); } public Task SortedSetAddAsync(RedisKey key, RedisValue member, double score, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => + return idempotentPolicy.ExecuteAsync(_ => UnderlyingDatabase.SortedSetAddAsync(key, member, score, flags), new Context()); } + public Task SortedSetUpdateAsync(RedisKey key, RedisValue member, double score, + SortedSetWhen when = SortedSetWhen.Always, CommandFlags flags = CommandFlags.None) + { + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.SortedSetUpdateAsync(key, member, score, when, flags), + new Context()); + } + + public Task SortedSetRemoveAsync(RedisKey key, RedisValue member, CommandFlags flags = CommandFlags.None) + { + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.SortedSetRemoveAsync(key, member, flags), + new Context()); + } + + public Task SortedSetRemoveAsync(RedisKey key, RedisValue[] members, CommandFlags flags = CommandFlags.None) + { + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.SortedSetRemoveAsync(key, members, flags), + new Context()); + } + public Task SortedSetRangeByRankWithScoresAsync(RedisKey key, long start = 0, long stop = -1, Order order = Order.Ascending, CommandFlags flags = CommandFlags.None) { - return policy.ExecuteAsync(_ => + return idempotentPolicy.ExecuteAsync(_ => UnderlyingDatabase.SortedSetRangeByRankWithScoresAsync(key, start, stop, order, flags), new Context()); } - public ITransaction CreateTransaction(object? asyncState = null) + public Task SortedSetRangeByScoreWithScoresAsync(RedisKey key, + double start = Double.NegativeInfinity, double stop = Double.PositiveInfinity, + Exclude exclude = Exclude.None, Order order = Order.Ascending, + long skip = 0, long take = -1, CommandFlags flags = CommandFlags.None) { - return UnderlyingDatabase.CreateTransaction(asyncState); - } - - public Task ScriptEvaluateAsync(string script, RedisKey[]? keys = null, - RedisValue[]? values = null, CommandFlags flags = CommandFlags.None) - { - return policy.ExecuteAsync(_ => - UnderlyingDatabase.ScriptEvaluateAsync(script, keys, values, flags), + return idempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.SortedSetRangeByScoreWithScoresAsync(key, start, stop, exclude, order, skip, take, flags), new Context()); } - public bool StringSet(RedisKey key, RedisValue value, TimeSpan? expiry = null, - When when = When.Always, CommandFlags flags = CommandFlags.None) + public Task PublishAsync(RedisChannel channel, RedisValue message, + CommandFlags flags = CommandFlags.None) { - return StringSetAsync(key, value, expiry, when, flags).GetAwaiter().GetResult(); - } - - public RedisValue StringGet(RedisKey key, CommandFlags flags = CommandFlags.None) - { - return StringGetAsync(key, flags).GetAwaiter().GetResult(); + return nonIdempotentPolicy.ExecuteAsync(_ => + UnderlyingDatabase.PublishAsync(channel, message, flags), + new Context()); } } }