Merge pull request #627 from imperugo/fix/pool-resilience-and-pubsub-handler

Fix pool resilience and PubSub silent exception swallowing

Author: imperugo

src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisClient.cs

@@ -1,5 +1,7 @@
 // Copyright (c) Ugo Lattanzi.  All Rights Reserved.  Licensed under the MIT license.  See License.txt in the project root for license information.
 
+using Microsoft.Extensions.Logging;
+
 using StackExchange.Redis.Extensions.Core.Abstractions;
 using StackExchange.Redis.Extensions.Core.Configuration;
 
@@ -9,6 +11,7 @@ namespace StackExchange.Redis.Extensions.Core.Implementations;
 public class RedisClient : IRedisClient
 {
     private readonly RedisConfiguration redisConfiguration;
+    private readonly ILogger<RedisDatabase>? databaseLogger;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="RedisClient"/> class.
@@ -24,6 +27,7 @@ public RedisClient(
         ConnectionPoolManager = connectionPoolManager;
         Serializer = serializer;
         this.redisConfiguration = redisConfiguration;
+        databaseLogger = redisConfiguration.LoggerFactory?.CreateLogger<RedisDatabase>();
     }
 
     /// <inheritdoc/>
@@ -92,7 +96,8 @@ public IRedisDatabase GetDb(int dbNumber, string? keyPrefix = null)
             redisConfiguration.ServerEnumerationStrategy,
             dbNumber,
             redisConfiguration.MaxValueLength,
-            keyPrefix);
+            keyPrefix,
+            databaseLogger);
     }
 
     /// <inheritdoc/>

src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisConnectionPoolManager.cs

@@ -109,16 +109,45 @@ var nextIdx
                 = new Random().Next(0, redisConfiguration.PoolSize);
 #endif
                 connection = connections[nextIdx];
+
+                if (!connection.IsConnected())
+                {
+                    // Selected connection is disconnected, try to find a connected one
+                    for (var i = 0; i < connections.Length; i++)
+                    {
+                        if (connections[i].IsConnected())
+                        {
+                            connection = connections[i];
+                            break;
+                        }
+                    }
+                }
+
                 break;
 
             case ConnectionSelectionStrategy.LeastLoaded:
-                connection = connections.MinBy(x => x.TotalOutstanding());
+                // Prefer connected connections; fall back to any if all are disconnected
+                IStateAwareConnection? candidate = null;
+
+                for (var i = 0; i < connections.Length; i++)
+                {
+                    if (!connections[i].IsConnected())
+                        continue;
+
+                    if (candidate == null || connections[i].TotalOutstanding() < candidate.TotalOutstanding())
+                        candidate = connections[i];
+                }
+
+                connection = candidate ?? connections.MinBy(x => x.TotalOutstanding());
                 break;
 
             default:
                 throw new InvalidEnumArgumentException(nameof(redisConfiguration.ConnectionSelectionStrategy), (int)redisConfiguration.ConnectionSelectionStrategy, typeof(ConnectionSelectionStrategy));
         }
 
+        if (!connection.IsConnected() && logger.IsEnabled(LogLevel.Warning))
+            logger.LogWarning("All Redis connections are disconnected. Using connection {HashCode} in degraded mode.", connection.Connection.GetHashCode().ToString(CultureInfo.InvariantCulture));
+
         if (logger.IsEnabled(LogLevel.Debug))
             logger.LogDebug("Using connection {HashCode} with {OutStanding} outstanding!", connection.Connection.GetHashCode().ToString(CultureInfo.InvariantCulture), connection.TotalOutstanding().ToString(CultureInfo.InvariantCulture));
 

src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.PubSub.cs

@@ -2,8 +2,11 @@
 
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.Extensions.Logging;
+
 using StackExchange.Redis.Extensions.Core.Helpers;
 
 namespace StackExchange.Redis.Extensions.Core.Implementations;
@@ -31,8 +34,20 @@ public Task SubscribeAsync<T>(RedisChannel channel, Func<T?, Task> handler, Comm
 
         return sub.SubscribeAsync(channel, Handler, flag);
 
-        void Handler(RedisChannel redisChannel, RedisValue value) =>
-            _ = handler(Serializer.Deserialize<T>(value)).ConfigureAwait(false);
+        void Handler(RedisChannel redisChannel, RedisValue value)
+        {
+            var task = Task.Run(async () =>
+            {
+                var deserialized = Serializer.Deserialize<T>(value);
+                await handler(deserialized).ConfigureAwait(false);
+            });
+
+            task.ContinueWith(
+                t => logger.LogError(t.Exception, "Error processing subscription message on channel {Channel}", (string?)redisChannel),
+                CancellationToken.None,
+                TaskContinuationOptions.OnlyOnFaulted,
+                TaskScheduler.Default);
+        }
     }
 
     /// <inheritdoc/>

src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.cs

@@ -8,6 +8,9 @@
 using System.Runtime.InteropServices;
 using System.Threading.Tasks;
 
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+
 using StackExchange.Redis.Extensions.Core.Abstractions;
 using StackExchange.Redis.Extensions.Core.Configuration;
 using StackExchange.Redis.Extensions.Core.Extensions;
@@ -26,6 +29,7 @@ public partial class RedisDatabase : IRedisDatabase
     private readonly string keyPrefix;
     private readonly uint maxValueLength;
     private readonly int dbNumber;
+    private readonly ILogger logger;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="RedisDatabase"/> class.
@@ -36,20 +40,23 @@ public partial class RedisDatabase : IRedisDatabase
     /// <param name="dbNumber">The database to use.</param>
     /// <param name="maxvalueLength">The max lenght of the cache object.</param>
     /// <param name="keyPrefix">The key prefix.</param>
+    /// <param name="logger">The logger.</param>
     public RedisDatabase(
         IRedisConnectionPoolManager connectionPoolManager,
         ISerializer serializer,
         ServerEnumerationStrategy serverEnumerationStrategy,
         int dbNumber,
         uint maxvalueLength,
-        string? keyPrefix = null)
+        string? keyPrefix = null,
+        ILogger<RedisDatabase>? logger = null)
     {
         Serializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
         this.connectionPoolManager = connectionPoolManager ?? throw new ArgumentNullException(nameof(connectionPoolManager));
         this.serverEnumerationStrategy = serverEnumerationStrategy;
         this.dbNumber = dbNumber;
         this.keyPrefix = keyPrefix ?? string.Empty;
         maxValueLength = maxvalueLength;
+        this.logger = logger ?? (ILogger)NullLogger<RedisDatabase>.Instance;
     }
 
     /// <inheritdoc/>

tests/StackExchange.Redis.Extensions.Core.Tests/CacheClientTestBase.PubSub.cs

@@ -0,0 +1,80 @@
+// Copyright (c) Ugo Lattanzi.  All Rights Reserved.  Licensed under the MIT license.  See License.txt in the project root for license information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Xunit;
+
+namespace StackExchange.Redis.Extensions.Core.Tests;
+
+public abstract partial class CacheClientTestBase
+{
+    [Fact]
+    public async Task SubscribeAsync_ValidMessage_ShouldInvokeHandler_Async()
+    {
+        var channel = new RedisChannel(Guid.NewGuid().ToString(), RedisChannel.PatternMode.Literal);
+        var received = new TaskCompletionSource<string?>();
+
+        await Sut.GetDefaultDatabase().SubscribeAsync<string>(channel, msg =>
+        {
+            received.TrySetResult(msg);
+            return Task.CompletedTask;
+        });
+
+        await Sut.GetDefaultDatabase().PublishAsync(channel, "hello");
+
+        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+        cts.Token.Register(() => received.TrySetCanceled());
+
+        var result = await received.Task;
+
+        Assert.Equal("hello", result);
+
+        await Sut.GetDefaultDatabase().UnsubscribeAllAsync();
+    }
+
+    [Fact]
+    public async Task SubscribeAsync_HandlerThrows_ShouldNotCrashProcess_Async()
+    {
+        var channel = new RedisChannel(Guid.NewGuid().ToString(), RedisChannel.PatternMode.Literal);
+        var secondMessageReceived = new TaskCompletionSource<bool>();
+        var firstHandlerCompleted = new TaskCompletionSource<bool>();
+        var callCount = 0;
+
+        await Sut.GetDefaultDatabase().SubscribeAsync<string>(channel, _ =>
+        {
+            var count = Interlocked.Increment(ref callCount);
+
+            if (count == 1)
+            {
+                firstHandlerCompleted.TrySetResult(true);
+                throw new InvalidOperationException("Simulated handler failure");
+            }
+
+            secondMessageReceived.TrySetResult(true);
+            return Task.CompletedTask;
+        });
+
+        // First message — handler throws, should be logged not crash
+        await Sut.GetDefaultDatabase().PublishAsync(channel, "msg1");
+
+        // Wait for first handler to complete before sending second message
+        using var cts1 = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+        cts1.Token.Register(() => firstHandlerCompleted.TrySetCanceled());
+        await firstHandlerCompleted.Task;
+
+        // Second message — handler should still be active
+        await Sut.GetDefaultDatabase().PublishAsync(channel, "msg2");
+
+        using var cts2 = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+        cts2.Token.Register(() => secondMessageReceived.TrySetCanceled());
+
+        var result = await secondMessageReceived.Task;
+
+        Assert.True(result, "Second message should have been received even after first handler threw");
+        Assert.True(callCount >= 2);
+
+        await Sut.GetDefaultDatabase().UnsubscribeAllAsync();
+    }
+}