From 1229f31a8ae9e13fbe1751b600a9e38f7ee43e4d Mon Sep 17 00:00:00 2001 From: rebaseandpanic Date: Tue, 14 Jul 2026 16:28:37 +0300 Subject: [PATCH] fix(RedisStreams): recover connection pool after a failed connect instead of caching it forever MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AsyncLazyRedisConnection derived from Lazy>, which caches the connect Task permanently. When the initial connect faulted — the Redis Sentinel path (GetSentinelMasterConnection) throws, unlike a normal abortConnect=false connect — the faulted Task was cached forever: every subsequent access rethrew the same exception and the factory was never retried. Worse, a single poisoned slot took the whole pool down, because connection selection dereferenced the throwing CreatedConnection getter for every slot. The pool is a singleton and its slots are created once, so the transport stayed dead until the pod restarted while the process kept running — a silent failure with no crash or alert. Reproduces only on CAP + Redis Sentinel, which is why it never surfaced upstream. Fix: - Drop the Lazy> base; hold the connect Task under a lock and recreate it when it faulted or was cancelled, reusing a healthy or in-flight Task otherwise. - Replace the throwing CreatedConnection getter with a non-throwing TryGetCreatedConnection(out ...); pool selection and Dispose skip poisoned slots so one bad slot can no longer take the pool down. - In the connect factory, dispose the multiplexer of every failed attempt and throw after exhausting retries, so the Task faults and the slot is recreated on next access — instead of caching a disconnected multiplexer as "healthy" forever. - Make pool disposal atomic (Interlocked) and clean up in-flight connects. This incidentally fixes two latent defects in the old retry loop: a disconnected multiplexer returned as success after exhausting attempts, and up to four leaked multiplexers per cycle. Adds DotNetCore.CAP.RedisStreams.Test covering recovery after failover, no-cache of a disconnected multiplexer, per-attempt disposal, poisoned-slot isolation, and pool disposal. BREAKING CHANGE: AsyncLazyRedisConnection no longer derives from Lazy> and no longer exposes CreatedConnection. It is internal transport plumbing; consider making it internal. --- CAP.sln | 7 + .../IConnectionPool.Default.cs | 158 +++++-- .../IConnectionPool.LazyConnection.cs | 183 ++++++-- .../DotNetCore.CAP.RedisStreams.Test.csproj | 28 ++ .../RedisConnectionPoolTest.cs | 444 ++++++++++++++++++ 5 files changed, 749 insertions(+), 71 deletions(-) create mode 100644 test/DotNetCore.CAP.RedisStreams.Test/DotNetCore.CAP.RedisStreams.Test.csproj create mode 100644 test/DotNetCore.CAP.RedisStreams.Test/RedisConnectionPoolTest.cs diff --git a/CAP.sln b/CAP.sln index 911d56f0a..7c3457a9a 100644 --- a/CAP.sln +++ b/CAP.sln @@ -60,6 +60,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotNetCore.CAP.InMemoryStor EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotNetCore.CAP.Test", "test\DotNetCore.CAP.Test\DotNetCore.CAP.Test.csproj", "{75CC45E6-BF06-40F4-977D-10DCC05B2EFA}" EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotNetCore.CAP.RedisStreams.Test", "test\DotNetCore.CAP.RedisStreams.Test\DotNetCore.CAP.RedisStreams.Test.csproj", "{37974F04-31B4-4B2F-826A-CD61D4680054}" +EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample.ConsoleApp", "samples\Sample.ConsoleApp\Sample.ConsoleApp.csproj", "{2B0F467E-ABBD-4A51-BF38-D4F609DB6266}" EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotNetCore.CAP.AmazonSQS", "src\DotNetCore.CAP.AmazonSQS\DotNetCore.CAP.AmazonSQS.csproj", "{43475E00-51B7-443D-BC2D-FC21F9D8A0B4}" @@ -160,6 +162,10 @@ Global {75CC45E6-BF06-40F4-977D-10DCC05B2EFA}.Debug|Any CPU.Build.0 = Debug|Any CPU {75CC45E6-BF06-40F4-977D-10DCC05B2EFA}.Release|Any CPU.ActiveCfg = Release|Any CPU {75CC45E6-BF06-40F4-977D-10DCC05B2EFA}.Release|Any CPU.Build.0 = Release|Any CPU + {37974F04-31B4-4B2F-826A-CD61D4680054}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {37974F04-31B4-4B2F-826A-CD61D4680054}.Debug|Any CPU.Build.0 = Debug|Any CPU + {37974F04-31B4-4B2F-826A-CD61D4680054}.Release|Any CPU.ActiveCfg = Release|Any CPU + {37974F04-31B4-4B2F-826A-CD61D4680054}.Release|Any CPU.Build.0 = Release|Any CPU {2B0F467E-ABBD-4A51-BF38-D4F609DB6266}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {2B0F467E-ABBD-4A51-BF38-D4F609DB6266}.Debug|Any CPU.Build.0 = Debug|Any CPU {2B0F467E-ABBD-4A51-BF38-D4F609DB6266}.Release|Any CPU.ActiveCfg = Release|Any CPU @@ -234,6 +240,7 @@ Global {F8EF381A-FE83-40B3-A63D-09D83851B0FB} = {10C0818D-9160-4B80-BB86-DDE925B64D43} {93176BAE-914B-4BED-9DE3-01FFB4F27FC5} = {9B2AE124-6636-4DE9-83A3-70360DABD0C4} {75CC45E6-BF06-40F4-977D-10DCC05B2EFA} = {C09CDAB0-6DD4-46E9-B7F3-3EF2A4741EA0} + {37974F04-31B4-4B2F-826A-CD61D4680054} = {C09CDAB0-6DD4-46E9-B7F3-3EF2A4741EA0} {2B0F467E-ABBD-4A51-BF38-D4F609DB6266} = {3A6B6931-A123-477A-9469-8B468B5385AF} {43475E00-51B7-443D-BC2D-FC21F9D8A0B4} = {9B2AE124-6636-4DE9-83A3-70360DABD0C4} {8B2FD3EA-E72B-4A82-B182-B87EC0C15D07} = {9B2AE124-6636-4DE9-83A3-70360DABD0C4} diff --git a/src/DotNetCore.CAP.RedisStreams/IConnectionPool.Default.cs b/src/DotNetCore.CAP.RedisStreams/IConnectionPool.Default.cs index 401c86910..3c9e50c71 100644 --- a/src/DotNetCore.CAP.RedisStreams/IConnectionPool.Default.cs +++ b/src/DotNetCore.CAP.RedisStreams/IConnectionPool.Default.cs @@ -1,8 +1,9 @@ -// Copyright (c) .NET Core Community. All rights reserved. +// Copyright (c) .NET Core Community. All rights reserved. // Licensed under the MIT License. See License.txt in the project root for license information. using System; using System.Collections.Concurrent; +using System.IO; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -16,26 +17,32 @@ internal class RedisConnectionPool : IRedisConnectionPool, IDisposable { private readonly ConcurrentBag _connections = []; + private readonly Func>? _connectionFactory; private readonly ILoggerFactory _loggerFactory; - private readonly SemaphoreSlim _poolLock = new(1); private readonly CapRedisOptions _redisOptions; - private bool _isDisposed; - private bool _poolAlreadyConfigured; + private readonly TimeSpan? _retryDelay; + private int _disposed; public RedisConnectionPool(IOptions options, ILoggerFactory loggerFactory) + : this(options, loggerFactory, connectionFactory: null, retryDelay: null) { - _redisOptions = options.Value; - _loggerFactory = loggerFactory; - Init().GetAwaiter().GetResult(); } - private AsyncLazyRedisConnection? QuietConnection + /// + /// Testing seam. Lets unit tests inject a fake multiplexer factory and a shorter + /// retry delay into every slot the pool creates; null means "use the real + /// defaults" (see ). DI always resolves the + /// public constructor, so production behavior is unchanged. + /// + internal RedisConnectionPool(IOptions options, ILoggerFactory loggerFactory, + Func>? connectionFactory, + TimeSpan? retryDelay) { - get - { - return _poolAlreadyConfigured ? _connections.OrderBy(static c => c.CreatedConnection?.ConnectionCapacity ?? int.MaxValue) - .First() : null; - } + _redisOptions = options.Value; + _loggerFactory = loggerFactory; + _connectionFactory = connectionFactory; + _retryDelay = retryDelay; + Init(); } public void Dispose() @@ -46,57 +53,124 @@ public void Dispose() public async Task ConnectAsync() { - if (QuietConnection == null) + if (Volatile.Read(ref _disposed) != 0) throw new ObjectDisposedException(nameof(RedisConnectionPool)); + + // Recompute pool state from healthy slots only. A slot whose connect task + // faulted (e.g. sentinel failover) must NOT count as "configured", otherwise + // the pool would consider itself full and keep handing out the poisoned slot. + // Kept as a local — persisting it as a shared field caused a data race under + // concurrent publish/consume. + var poolConfigured = + _connections.Count(static c => c.TryGetCreatedConnection(out _)) == _redisOptions.ConnectionPoolSize; + + if (poolConfigured) { - _poolAlreadyConfigured = - _connections.Count(static c => c.IsValueCreated) == _redisOptions.ConnectionPoolSize; - if (QuietConnection != null) return QuietConnection.CreatedConnection!.Connection; + var quiet = SelectLeastLoadedHealthy(); + if (quiet != null) return quiet.Connection; } foreach (var lazy in _connections) { - if (!lazy.IsValueCreated) return (await lazy).Connection; + // Uninitialized or poisoned slot → (re)connect it. A healthy in-flight + // task is reused by GetAwaiter; a faulted/cancelled one is discarded and + // the factory retried. + if (!lazy.TryGetCreatedConnection(out var created)) return (await lazy).Connection; - if (lazy.CreatedConnection!.ConnectionCapacity == default) return lazy.CreatedConnection.Connection; + // Healthy connection with no outstanding operations → reuse it. + if (created.ConnectionCapacity == default) return created.Connection; } - return (await _connections.OrderBy(static c => c.CreatedConnection!.ConnectionCapacity).First()).Connection; + // All slots are healthy and busy → pick the least-loaded one. + var leastLoaded = SelectLeastLoadedHealthy(); + if (leastLoaded != null) return leastLoaded.Connection; + + // Unreachable with a non-empty pool: a successfully completed task is terminal, + // so a slot the loop above saw as healthy cannot turn unhealthy before + // SelectLeastLoadedHealthy runs. Kept as a defensive last resort; on an empty + // pool First() throws, but PostConfigure guarantees at least one slot. + return (await _connections.First()).Connection; } - private async Task Init() + /// + /// Returns the healthy connection with the lowest outstanding capacity, skipping + /// any poisoned slot. Never throws — used by both the hot path and diagnostics. + /// + private RedisConnection? SelectLeastLoadedHealthy() { - try - { - await _poolLock.WaitAsync(); + RedisConnection? best = null; + var bestCapacity = long.MaxValue; - if (!_connections.IsEmpty) return; + foreach (var lazy in _connections) + { + if (!lazy.TryGetCreatedConnection(out var created)) continue; - for (var i = 0; i < _redisOptions.ConnectionPoolSize; i++) + var capacity = created.ConnectionCapacity; + if (capacity < bestCapacity) { - var connection = new AsyncLazyRedisConnection(_redisOptions, - _loggerFactory.CreateLogger()); - - _connections.Add(connection); + bestCapacity = capacity; + best = created; } } - finally + + return best; + } + + private void Init() + { + if (!_connections.IsEmpty) return; + + for (var i = 0; i < _redisOptions.ConnectionPoolSize; i++) { - _poolLock.Release(); + var connection = new AsyncLazyRedisConnection(_redisOptions, + _loggerFactory.CreateLogger(), _connectionFactory, _retryDelay); + + _connections.Add(connection); } } + // Pool disposal is best-effort at shutdown: a ConnectAsync racing with Dispose may + // leak a multiplexer or observe a disposed connection. For an in-flight caller that + // already passed the disposed-check this is not just a leak but a use-after-dispose: + // it can be handed a connection that the cleanup continuation below disposes right + // away, so its next publish/consume fails with ObjectDisposedException. Accepted — + // the process is terminating and the OS reclaims sockets; fully closing the window + // would require locking the publish/consume hot path for the sake of a dying process. private void Dispose(bool disposing) { - if (_isDisposed) return; - - if (disposing) - foreach (var connection in _connections) - { - if (!connection.IsValueCreated) continue; + // Atomic check/set: a concurrent Dispose must not run the cleanup loop twice. + // Publishes disposal before releasing connections so a concurrent ConnectAsync + // observes it and does not create a fresh slot no one would ever close. + if (Interlocked.Exchange(ref _disposed, 1) != 0) return; - connection.CreatedConnection!.Dispose(); - } + if (!disposing) return; - _isDisposed = true; + foreach (var connection in _connections) + { + // Read the slot's task once, then dispose its connection through a single + // continuation. ContinueWith runs synchronously and immediately when the task + // is already complete, so already-connected, in-flight, and racing-completion + // are all handled with ONE observation of the task's state — branching on two + // separate reads (IsCompletedSuccessfully then IsCompleted) would let a task + // that completes between the reads slip through undisposed. Faulted/cancelled + // tasks produce no connection, so the success guard skips them. Don't block + // shutdown waiting for an in-flight connect to finish. + var pending = connection.PeekTask(); + if (pending is null) continue; // slot never initialized — nothing to dispose + + pending.ContinueWith(static t => + { + if (t.IsCompletedSuccessfully) + // Swallow failures so one slot's Dispose can't leave the rest leaked. + try + { + t.Result.Dispose(); + } + catch + { + /* best-effort cleanup at shutdown */ + } + }, + TaskContinuationOptions.ExecuteSynchronously); + } } -} \ No newline at end of file +} diff --git a/src/DotNetCore.CAP.RedisStreams/IConnectionPool.LazyConnection.cs b/src/DotNetCore.CAP.RedisStreams/IConnectionPool.LazyConnection.cs index 6f6e77de0..d93cfb06e 100644 --- a/src/DotNetCore.CAP.RedisStreams/IConnectionPool.LazyConnection.cs +++ b/src/DotNetCore.CAP.RedisStreams/IConnectionPool.LazyConnection.cs @@ -1,7 +1,9 @@ -// Copyright (c) .NET Core Community. All rights reserved. +// Copyright (c) .NET Core Community. All rights reserved. // Licensed under the MIT License. See License.txt in the project root for license information. using System; +using System.Diagnostics.CodeAnalysis; +using System.IO; using System.Runtime.CompilerServices; using System.Threading.Tasks; using Microsoft.Extensions.Logging; @@ -9,53 +11,176 @@ namespace DotNetCore.CAP.RedisStreams; -public class AsyncLazyRedisConnection( - CapRedisOptions redisOptions, - ILogger logger) - : Lazy>(() => ConnectAsync(redisOptions, logger)) +public class AsyncLazyRedisConnection { - public RedisConnection? CreatedConnection - => IsValueCreated ? Value.GetAwaiter().GetResult() : null; + internal const int MaxConnectAttempts = 5; + private static readonly TimeSpan DefaultRetryDelay = TimeSpan.FromSeconds(2); + + private readonly Func> _connectionFactory; + private readonly ILogger _logger; + private readonly CapRedisOptions _redisOptions; + private readonly TimeSpan _retryDelay; + + // Guards creation/replacement of the cached connect task. The previous + // implementation inherited from Lazy>, which caches a + // faulted task forever: once the sentinel connect path threw, the pool never + // recovered after a Redis failover. We keep the task ourselves so a faulted or + // cancelled task can be discarded and the factory retried, while a healthy or + // still in-flight task is reused. + private readonly object _lock = new(); + private Task? _task; + + public AsyncLazyRedisConnection( + CapRedisOptions redisOptions, + ILogger logger) + : this(redisOptions, logger, connectionFactory: null, retryDelay: null) + { + } - public TaskAwaiter GetAwaiter() + /// + /// Testing seam. Lets unit tests substitute the multiplexer factory (so the + /// retry/dispose/recovery logic can run against a fake instead of a real network + /// connect) and shorten the delay between attempts. Production code always goes + /// through the public constructor, which keeps the real + /// + /// factory and the default retry delay. + /// + internal AsyncLazyRedisConnection( + CapRedisOptions redisOptions, + ILogger logger, + Func>? connectionFactory, + TimeSpan? retryDelay) { - return Value.GetAwaiter(); + _redisOptions = redisOptions; + _logger = logger; + _connectionFactory = connectionFactory ?? DefaultConnectionFactoryAsync; + _retryDelay = retryDelay ?? DefaultRetryDelay; } - private static async Task ConnectAsync(CapRedisOptions redisOptions, - ILogger logger) + /// + /// Non-blocking, non-throwing accessor for the created connection. Returns + /// false for a slot that has not been created, is still connecting, or + /// whose connect task faulted or was cancelled (a "poisoned" slot). + /// + public bool TryGetCreatedConnection([NotNullWhen(true)] out RedisConnection? connection) { - var attempt = 1; + Task? task; + lock (_lock) + { + task = _task; + } - var redisLogger = new RedisLogger(logger); + if (task is { IsCompletedSuccessfully: true }) + { + connection = task.Result; + return true; + } - ConnectionMultiplexer? connection = null; + connection = null; + return false; + } - while (attempt <= 5) + /// + /// Returns the current connect task without creating or replacing it. Unlike + /// this exposes an in-flight or poisoned + /// task too, so pool disposal can attach cleanup to a connection that is still + /// being established without forcing a reconnect. + /// + public Task? PeekTask() + { + lock (_lock) { - connection = await ConnectionMultiplexer.ConnectAsync(redisOptions.Configuration!, redisLogger) - .ConfigureAwait(false); + return _task; + } + } - connection.LogEvents(logger); + public TaskAwaiter GetAwaiter() + { + return GetOrCreateTask().GetAwaiter(); + } - if (!connection.IsConnected) - { - logger.LogWarning("Can't establish redis connection,trying to establish connection [attempt {attempt}].", attempt); + /// + /// Returns the cached connect task, (re)creating it when the slot has never been + /// initialized or when the previous task faulted or was cancelled. A healthy or + /// still in-flight task is reused. Racing callers are serialized by the lock, so + /// the factory is invoked at most once per reset. + /// + private Task GetOrCreateTask() + { + lock (_lock) + { + if (_task is null || _task.IsFaulted || _task.IsCanceled) + _task = ConnectAsync(); + + return _task; + } + } - await Task.Delay(TimeSpan.FromSeconds(2)) + private static async Task DefaultConnectionFactoryAsync( + ConfigurationOptions configuration, TextWriter log) + { + return await ConnectionMultiplexer.ConnectAsync(configuration, log).ConfigureAwait(false); + } + + private async Task ConnectAsync() + { + var redisLogger = new RedisLogger(_logger); + + Exception? lastException = null; + + for (var attempt = 1; attempt <= MaxConnectAttempts; attempt++) + { + IConnectionMultiplexer? connection = null; + try + { + connection = await _connectionFactory(_redisOptions.Configuration!, redisLogger) .ConfigureAwait(false); - ++attempt; + connection.LogEvents(_logger); + + if (connection.IsConnected) + return new RedisConnection(connection); + + // AbortOnConnectFail=false returns a DISCONNECTED multiplexer instead of + // throwing. Treat it as a failed attempt so we never cache a dead connection. + lastException = new InvalidOperationException($"Redis connection is not connected [attempt {attempt}]."); } - else + catch (Exception ex) { - attempt = 6; + // AbortOnConnectFail=true makes ConnectAsync throw once connect fails. + lastException = ex; } - } - if (connection == null) throw new Exception($"Can't establish redis connection,after [{attempt}] attempts."); + // Failed attempt: dispose the multiplexer (if one was created) so its sockets + // and background reconnect loop don't leak, then retry after a delay. This runs + // on both the throw path and the disconnected-multiplexer path. Swallow a + // failing Dispose so it can't abort the remaining retry attempts. + try + { + connection?.Dispose(); + } + catch + { + /* best-effort cleanup between retries */ + } + + // Only advertise a follow-up attempt when one will actually happen. On the + // final failure we stay silent here; the method then throws with a summarizing + // message so we don't promise a retry that never comes. + if (attempt < MaxConnectAttempts) + { + _logger.LogWarning(lastException, + "Can't establish redis connection, trying to establish connection [attempt {attempt}].", attempt); - return new RedisConnection(connection); + await Task.Delay(_retryDelay).ConfigureAwait(false); + } + } + + // All attempts exhausted without a live connection. Throw so the connect task + // faults and GetOrCreateTask recreates the slot on the next access, instead of + // caching a dead/disconnected multiplexer forever. + throw new InvalidOperationException( + $"Can't establish redis connection after [{MaxConnectAttempts}] attempts.", lastException); } } @@ -79,4 +204,4 @@ private void Dispose(bool disposing) _isDisposed = true; } -} \ No newline at end of file +} diff --git a/test/DotNetCore.CAP.RedisStreams.Test/DotNetCore.CAP.RedisStreams.Test.csproj b/test/DotNetCore.CAP.RedisStreams.Test/DotNetCore.CAP.RedisStreams.Test.csproj new file mode 100644 index 000000000..6160daf98 --- /dev/null +++ b/test/DotNetCore.CAP.RedisStreams.Test/DotNetCore.CAP.RedisStreams.Test.csproj @@ -0,0 +1,28 @@ + + + + net10.0 + enable + + false + + + + + + + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + + + + diff --git a/test/DotNetCore.CAP.RedisStreams.Test/RedisConnectionPoolTest.cs b/test/DotNetCore.CAP.RedisStreams.Test/RedisConnectionPoolTest.cs new file mode 100644 index 000000000..067694c39 --- /dev/null +++ b/test/DotNetCore.CAP.RedisStreams.Test/RedisConnectionPoolTest.cs @@ -0,0 +1,444 @@ +// Copyright (c) .NET Core Community. All rights reserved. +// Licensed under the MIT License. See License.txt in the project root for license information. + +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using StackExchange.Redis; +using Xunit; + +namespace DotNetCore.CAP.RedisStreams.Test; + +/// +/// Covers the connection pool recovery fix: a slot whose connect attempt failed +/// (threw, or produced a disconnected multiplexer) must not be cached forever. +/// The pool has to retry the factory and hand out a live connection once Redis +/// is reachable again, and every failed-attempt multiplexer must be disposed. +/// +public class RedisConnectionPoolTest +{ + [Fact] + public async Task ConnectAsync_recovers_after_redis_comes_back() + { + // First run: every attempt throws (e.g. sentinel failover while the master is + // re-elected). Second run: Redis is back and the factory returns a connected + // multiplexer. + var connected = CreateMultiplexer(isConnected: true); + var factoryCalls = 0; + + Task Factory(ConfigurationOptions configuration, TextWriter log) + { + factoryCalls++; + if (factoryCalls <= AsyncLazyRedisConnection.MaxConnectAttempts) + return Task.FromException( + new RedisConnectionException(ConnectionFailureType.UnableToConnect, "sentinel failover")); + + return Task.FromResult(connected); + } + + using var pool = CreatePool(Factory); + + var thrown = await Assert.ThrowsAsync(() => pool.ConnectAsync()); + Assert.IsType(thrown.InnerException); + Assert.Equal(AsyncLazyRedisConnection.MaxConnectAttempts, factoryCalls); + + // The old Lazy>-based slot cached the faulted task forever, so this + // second call used to rethrow the stale exception instead of reconnecting. + var result = await pool.ConnectAsync(); + + Assert.Same(connected, result); + Assert.Equal(AsyncLazyRedisConnection.MaxConnectAttempts + 1, factoryCalls); + } + + [Fact] + public async Task ConnectAsync_does_not_cache_disconnected_multiplexer_as_healthy() + { + // AbortOnConnectFail=false makes ConnectionMultiplexer.ConnectAsync return a + // DISCONNECTED multiplexer instead of throwing. The pool must treat it as a + // failed attempt, keep calling the factory on later requests, and never hand + // the dead multiplexer out. + var disconnected = new List(); + var connected = CreateMultiplexer(isConnected: true); + var redisIsUp = false; + + Task Factory(ConfigurationOptions configuration, TextWriter log) + { + if (redisIsUp) return Task.FromResult(connected); + + var multiplexer = CreateMultiplexer(isConnected: false); + disconnected.Add(multiplexer); + return Task.FromResult(multiplexer); + } + + using var pool = CreatePool(Factory); + + await Assert.ThrowsAsync(() => pool.ConnectAsync()); + Assert.Equal(AsyncLazyRedisConnection.MaxConnectAttempts, disconnected.Count); + + // The slot is poisoned, not healthy: the next request runs the factory again + // (a full second round of attempts) instead of returning a dead connection. + await Assert.ThrowsAsync(() => pool.ConnectAsync()); + Assert.Equal(2 * AsyncLazyRedisConnection.MaxConnectAttempts, disconnected.Count); + + redisIsUp = true; + var result = await pool.ConnectAsync(); + + Assert.Same(connected, result); + } + + [Fact] + public async Task ConnectAsync_disposes_every_failed_attempt_multiplexer() + { + // Each failed attempt creates a multiplexer with live sockets and a background + // reconnect loop; leaving them undisposed leaks both. Every one of the + // MaxConnectAttempts multiplexers must be disposed exactly once. + var disconnected = new List(); + + Task Factory(ConfigurationOptions configuration, TextWriter log) + { + var multiplexer = CreateMultiplexer(isConnected: false); + disconnected.Add(multiplexer); + return Task.FromResult(multiplexer); + } + + using var pool = CreatePool(Factory); + + await Assert.ThrowsAsync(() => pool.ConnectAsync()); + + Assert.Equal(AsyncLazyRedisConnection.MaxConnectAttempts, disconnected.Count); + foreach (var multiplexer in disconnected) + multiplexer.Received(1).Dispose(); + } + + [Fact] + public async Task ConnectAsync_returns_connected_multiplexer_on_first_attempt() + { + var connected = CreateMultiplexer(isConnected: true); + var factoryCalls = 0; + + Task Factory(ConfigurationOptions configuration, TextWriter log) + { + factoryCalls++; + return Task.FromResult(connected); + } + + using var pool = CreatePool(Factory); + + var result = await pool.ConnectAsync(); + + Assert.Same(connected, result); + Assert.Equal(1, factoryCalls); + connected.DidNotReceive().Dispose(); + + // A healthy cached slot is reused without another factory run. + var again = await pool.ConnectAsync(); + + Assert.Same(connected, again); + Assert.Equal(1, factoryCalls); + } + + [Fact] + public async Task Dispose_disposes_created_connections() + { + var connected = CreateMultiplexer(isConnected: true); + var pool = CreatePool((_, _) => Task.FromResult(connected)); + + await pool.ConnectAsync(); + + pool.Dispose(); + + connected.Received(1).Dispose(); + } + + [Fact] + public async Task ConnectAsync_throws_ObjectDisposedException_after_dispose() + { + var pool = CreatePool((_, _) => Task.FromResult(CreateMultiplexer(isConnected: true))); + pool.Dispose(); + + await Assert.ThrowsAsync(() => pool.ConnectAsync()); + } + + [Fact] + public async Task ConnectAsync_poisoned_slot_does_not_prevent_returning_healthy_connection() + { + // Core of the bug this fork fixes. In the old Lazy>-based pool the + // CreatedConnection getter did Value.GetAwaiter().GetResult(), so a slot whose + // connect task had faulted RETHREW that cached exception from every selection + // path (QuietConnection's OrderBy over all slots and the foreach capacity + // probe). One poisoned slot therefore took ConnectAsync down for the WHOLE + // pool - publish and consume both failed - even while another slot held a + // perfectly healthy connection, and the Lazy slot never retried. This test + // drives a two-slot pool into the "one slot poisoned" state and proves later + // requests still get a live connection instead of the stale exception. + var connected = CreateMultiplexer(isConnected: true); + var factoryCalls = 0; + + Task Factory(ConfigurationOptions configuration, TextWriter log) + { + factoryCalls++; + // Exactly one full retry round fails, poisoning the first slot the pool + // touches; every later call yields a healthy multiplexer. + if (factoryCalls <= AsyncLazyRedisConnection.MaxConnectAttempts) + return Task.FromException( + new RedisConnectionException(ConnectionFailureType.UnableToConnect, "node down")); + + return Task.FromResult(connected); + } + + using var pool = CreatePool(Factory, poolSize: 2); + + // Poison one slot: the first request exhausts all attempts on it and throws. + await Assert.ThrowsAsync(() => pool.ConnectAsync()); + Assert.Equal(AsyncLazyRedisConnection.MaxConnectAttempts, factoryCalls); + + // A poisoned slot now sits in the pool. The next request must NOT rethrow its + // cached failure: the pool (re)connects a non-healthy slot with the now-working + // factory - exactly one more run - and hands out the live multiplexer. + var result = await pool.ConnectAsync(); + + Assert.Same(connected, result); + Assert.Equal(AsyncLazyRedisConnection.MaxConnectAttempts + 1, factoryCalls); + + // Depending on ConcurrentBag enumeration order the call above healed either + // the poisoned slot (leaving the other uninitialized) or the untouched slot + // (leaving the poison in place). Both states mix one healthy slot with one + // non-healthy slot, and the next request must again return a live connection + // without throwing: it either reuses the healthy slot (no factory run) or + // repairs the broken one (one factory run). + var again = await pool.ConnectAsync(); + + Assert.Same(connected, again); + Assert.InRange(factoryCalls, + AsyncLazyRedisConnection.MaxConnectAttempts + 1, + AsyncLazyRedisConnection.MaxConnectAttempts + 2); + } + + [Fact] + public async Task TryGetCreatedConnection_does_not_throw_for_poisoned_slot() + { + // Slot-level half of the poisoned-slot invariant: the accessor the pool's + // selection paths rely on (the poolConfigured count and + // SelectLeastLoadedHealthy both call TryGetCreatedConnection) must stay + // non-throwing for a faulted slot so selection can skip it. The old + // equivalent, CreatedConnection => Value.GetAwaiter().GetResult(), rethrew + // the cached connect exception - that is what made one poisoned slot fatal + // pool-wide. + var poisonedSlot = CreateSlot((_, _) => Task.FromException( + new RedisConnectionException(ConnectionFailureType.UnableToConnect, "node down"))); + + await Assert.ThrowsAsync(async () => await poisonedSlot); + + // Non-throwing "no usable connection" answer lets selection skip the slot. + Assert.False(poisonedSlot.TryGetCreatedConnection(out var fromPoisoned)); + Assert.Null(fromPoisoned); + // The faulted task itself stays observable (pool disposal peeks at it). + Assert.True(poisonedSlot.PeekTask()!.IsFaulted); + + var connected = CreateMultiplexer(isConnected: true); + var healthySlot = CreateSlot((_, _) => Task.FromResult(connected)); + var healthy = await healthySlot; + + Assert.True(healthySlot.TryGetCreatedConnection(out var fromHealthy)); + Assert.Same(healthy, fromHealthy); + Assert.Same(connected, fromHealthy!.Connection); + } + + [Fact] + public async Task Dispose_disposes_multiplexer_from_connect_still_in_flight() + { + // Disposing the pool while a slot's connect attempt is still running must not + // leak the connection that attempt eventually produces: Dispose attaches a + // continuation to the in-flight task, and once the factory completes the + // produced multiplexer gets disposed. The factory is gated on a + // TaskCompletionSource so "in flight" is a guaranteed state, not a timing. + var connected = CreateMultiplexer(isConnected: true); + var gate = new TaskCompletionSource(); + + var pool = CreatePool((_, _) => gate.Task); + + var connectTask = pool.ConnectAsync(); + Assert.False(connectTask.IsCompleted); // connect is gated: still in flight + + pool.Dispose(); + connected.DidNotReceive().Dispose(); // nothing produced yet - nothing to dispose + + // Redis "answers" after shutdown started: the continuation registered by + // Dispose must clean the late connection up. + gate.SetResult(connected); + + // The request that had already passed the disposed-check still completes; + // pool disposal is documented as best-effort towards in-flight callers. + var result = await connectTask; + Assert.Same(connected, result); + + connected.Received(1).Dispose(); + } + + [Fact] + public async Task Dispose_called_twice_disposes_connection_only_once() + { + // Interlocked.Exchange in Dispose must make the cleanup run exactly once; + // RedisConnection's own dispose-guard is the second belt. The user-visible + // contract: double-disposing the pool neither throws nor disposes the + // multiplexer a second time. + var connected = CreateMultiplexer(isConnected: true); + var pool = CreatePool((_, _) => Task.FromResult(connected)); + + await pool.ConnectAsync(); + + pool.Dispose(); + pool.Dispose(); + + connected.Received(1).Dispose(); + } + + [Fact] + public void Dispose_with_never_connected_slot_does_not_connect_or_throw() + { + // A slot that was never asked for a connection has no task (PeekTask() is + // null). Disposing the pool must skip it silently - in particular it must + // not force a connect just to have something to dispose. + var factoryCalls = 0; + + Task Factory(ConfigurationOptions configuration, TextWriter log) + { + factoryCalls++; + return Task.FromResult(CreateMultiplexer(isConnected: true)); + } + + var pool = CreatePool(Factory); + + pool.Dispose(); + + Assert.Equal(0, factoryCalls); + } + + [Fact] + public async Task ConnectAsync_overlapping_requests_share_one_factory_run() + { + // GetOrCreateTask serializes racing callers on the slot lock: per reset the + // factory runs at most once and every caller awaits the SAME in-flight task. + // The factory is gated on a TaskCompletionSource, so each pool.ConnectAsync + // call below runs synchronously exactly up to the await on the slot's pending + // task - all eight requests verifiably overlap without any timing or sleeps. + var connected = CreateMultiplexer(isConnected: true); + var gate = new TaskCompletionSource(); + var factoryCalls = 0; + + Task Factory(ConfigurationOptions configuration, TextWriter log) + { + factoryCalls++; + return gate.Task; + } + + using var pool = CreatePool(Factory); + + var requests = Enumerable.Range(0, 8).Select(_ => pool.ConnectAsync()).ToArray(); + + Assert.Equal(1, factoryCalls); // one connect cycle despite eight waiters + Assert.All(requests, request => Assert.False(request.IsCompleted)); + + gate.SetResult(connected); + var results = await Task.WhenAll(requests); + + Assert.Equal(1, factoryCalls); + Assert.All(results, result => Assert.Same(connected, result)); + } + + [Fact] + public async Task ConnectAsync_keeps_retrying_when_failed_attempt_dispose_throws() + { + // The cleanup between attempts disposes the failed multiplexer inside a + // try/catch. If that guard were removed, the throwing Dispose below would abort + // the retry loop on the very first attempt and the slot could never recover. + // Every failed attempt here yields a disconnected multiplexer whose Dispose + // throws, and Redis "comes back" on the final attempt: the run must survive all + // the throwing Disposes and still hand out the live connection. + var connected = CreateMultiplexer(isConnected: true); + var disconnected = new List(); + var factoryCalls = 0; + + Task Factory(ConfigurationOptions configuration, TextWriter log) + { + factoryCalls++; + if (factoryCalls < AsyncLazyRedisConnection.MaxConnectAttempts) + { + var multiplexer = CreateMultiplexer(isConnected: false); + multiplexer.When(m => m.Dispose()).Do(_ => throw new InvalidOperationException("dispose failed")); + disconnected.Add(multiplexer); + return Task.FromResult(multiplexer); + } + + return Task.FromResult(connected); + } + + using var pool = CreatePool(Factory); + + var result = await pool.ConnectAsync(); + + Assert.Same(connected, result); + Assert.Equal(AsyncLazyRedisConnection.MaxConnectAttempts, factoryCalls); + + // Each failed-attempt multiplexer had Dispose attempted exactly once; the throw + // was swallowed instead of aborting the remaining attempts. + Assert.Equal(AsyncLazyRedisConnection.MaxConnectAttempts - 1, disconnected.Count); + foreach (var multiplexer in disconnected) + multiplexer.Received(1).Dispose(); + } + + /// + /// Pool with the injected factory, the given slot count (one by default) and a + /// zero retry delay so the MaxConnectAttempts retry loop completes without real + /// waiting. Production keeps the default 2-second delay via the public + /// constructors. + /// + private static RedisConnectionPool CreatePool( + Func> connectionFactory, + uint poolSize = 1) + { + return new RedisConnectionPool(Options.Create(CreateRedisOptions(poolSize)), NullLoggerFactory.Instance, + connectionFactory, TimeSpan.Zero); + } + + /// + /// Standalone slot with the injected factory and a zero retry delay, for tests + /// that exercise the slot's own accessors rather than the pool's selection. + /// + private static AsyncLazyRedisConnection CreateSlot( + Func> connectionFactory) + { + return new AsyncLazyRedisConnection(CreateRedisOptions(poolSize: 1), + NullLogger.Instance, connectionFactory, TimeSpan.Zero); + } + + private static CapRedisOptions CreateRedisOptions(uint poolSize) + { + return new CapRedisOptions + { + Configuration = ConfigurationOptions.Parse("localhost:6379"), + ConnectionPoolSize = poolSize + }; + } + + /// + /// Fake multiplexer with a fixed + /// state. Event subscription (used by LogEvents) works out of the box on + /// the substitute; returns + /// empty counters so the pool's least-loaded selection sees zero outstanding + /// operations. Dispose calls are counted by NSubstitute. + /// + private static IConnectionMultiplexer CreateMultiplexer(bool isConnected) + { + var multiplexer = Substitute.For(); + multiplexer.IsConnected.Returns(isConnected); + multiplexer.GetCounters().Returns(new ServerCounters(new DnsEndPoint("localhost", 6379))); + return multiplexer; + } +}