Skip to content

fix(RedisStreams): recover connection pool after a failed connect instead of caching it forever #1808

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rebaseandpanic wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(RedisStreams): recover connection pool after a failed connect instead of caching it forever #1808

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions CAP.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}"
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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}
Expand Down
158 changes: 116 additions & 42 deletions src/DotNetCore.CAP.RedisStreams/IConnectionPool.Default.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -16,26 +17,32 @@ internal class RedisConnectionPool : IRedisConnectionPool, IDisposable
{
private readonly ConcurrentBag<AsyncLazyRedisConnection> _connections = [];

private readonly Func<ConfigurationOptions, TextWriter, Task<IConnectionMultiplexer>>? _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<CapRedisOptions> options, ILoggerFactory loggerFactory)
: this(options, loggerFactory, connectionFactory: null, retryDelay: null)
{
_redisOptions = options.Value;
_loggerFactory = loggerFactory;
Init().GetAwaiter().GetResult();
}

private AsyncLazyRedisConnection? QuietConnection
/// <summary>
/// Testing seam. Lets unit tests inject a fake multiplexer factory and a shorter
/// retry delay into every slot the pool creates; <c>null</c> means "use the real
/// defaults" (see <see cref="AsyncLazyRedisConnection" />). DI always resolves the
/// public constructor, so production behavior is unchanged.
/// </summary>
internal RedisConnectionPool(IOptions<CapRedisOptions> options, ILoggerFactory loggerFactory,
Func<ConfigurationOptions, TextWriter, Task<IConnectionMultiplexer>>? 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()
Expand All @@ -46,57 +53,124 @@ public void Dispose()

public async Task<IConnectionMultiplexer> 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()
/// <summary>
/// Returns the healthy connection with the lowest outstanding capacity, skipping
/// any poisoned slot. Never throws — used by both the hot path and diagnostics.
/// </summary>
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<AsyncLazyRedisConnection>());

_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<AsyncLazyRedisConnection>(), _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);
}
}
}
}
Loading