From 81c97889ada67bd781e6eaf8171cace4ae0f6bd3 Mon Sep 17 00:00:00 2001 From: Harshal Patel Date: Tue, 23 Jun 2026 21:47:42 +0530 Subject: [PATCH 1/5] fix(bridge): enforce TaskCompletionSource handshake gate to prevent connection race --- src/StackExchange.Redis/PhysicalBridge.cs | 20 +++++- .../ConnectionRaceTests.cs | 72 +++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) create mode 100644 tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 21047efa2..7ce2572e5 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; @@ -56,6 +56,8 @@ internal sealed class PhysicalBridge : IDisposable private volatile bool reportNextFailure = true, reconfigureNextFailure = false; + private TaskCompletionSource _handshakeCompletion = new(TaskCreationOptions.RunContinuationsAsynchronously); + private volatile int state = (int)State.Disconnected; internal long? ConnectionId => physical?.ConnectionId; @@ -526,6 +528,7 @@ internal void OnFullyEstablished(PhysicalConnection connection, string source) connection.SetIdle(); if (physical == connection && !isDisposed && ChangeState(State.ConnectedEstablishing, State.ConnectedEstablished)) { + _handshakeCompletion.TrySetResult(true); reportNextFailure = reconfigureNextFailure = true; LastException = null; Interlocked.Exchange(ref failConnectCount, 0); @@ -800,6 +803,10 @@ private WriteResult WriteMessageInsideLock(PhysicalConnection physical, Message internal WriteResult WriteMessageTakingWriteLockSync(PhysicalConnection physical, Message message) { Trace("Writing: " + message); + if (ConnectionState == State.ConnectedEstablishing) + { + _handshakeCompletion.Task.GetAwaiter().GetResult(); + } message.SetEnqueued(physical); // this also records the read/write stats at this point // AVOID REORDERING MESSAGES @@ -1235,6 +1242,10 @@ private WriteResult TimedOutBeforeWrite(Message message) internal ValueTask WriteMessageTakingWriteLockAsync(PhysicalConnection physical, Message message, bool bypassBacklog = false) { Trace("Writing: " + message); + if (ConnectionState == State.ConnectedEstablishing && !_handshakeCompletion.Task.IsCompleted) + { + return WriteMessageTakingWriteLockAsync_Gate(physical, message, bypassBacklog); + } message.SetEnqueued(physical); // this also records the read/write stats at this point // AVOID REORDERING MESSAGES @@ -1296,6 +1307,12 @@ internal ValueTask WriteMessageTakingWriteLockAsync(PhysicalConnect } } + private async ValueTask WriteMessageTakingWriteLockAsync_Gate(PhysicalConnection physical, Message message, bool bypassBacklog) + { + await _handshakeCompletion.Task.ConfigureAwait(false); + return await WriteMessageTakingWriteLockAsync(physical, message, bypassBacklog).ConfigureAwait(false); + } + private async ValueTask WriteMessageTakingWriteLockAsync_Awaited( ValueTask pending, PhysicalConnection physical, @@ -1404,6 +1421,7 @@ private bool ChangeState(State oldState, State newState) Volatile.Write(ref _needsReconnect, false); Interlocked.Increment(ref socketCount); Interlocked.Exchange(ref connectStartTicks, Environment.TickCount); + Interlocked.Exchange(ref _handshakeCompletion, new(TaskCreationOptions.RunContinuationsAsynchronously)); // separate creation and connection for case when connection completes synchronously // in that case PhysicalConnection will call back to PhysicalBridge, and most PhysicalBridge methods assume that physical is not null; physical = new PhysicalConnection(this, Multiplexer.RawConfig.WriteMode); diff --git a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs new file mode 100644 index 000000000..2ab0a7306 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs @@ -0,0 +1,72 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Xunit; +using Xunit.Abstractions; + +namespace StackExchange.Redis.Tests; + +public class ConnectionRaceTests(ITestOutputHelper output) : TestBase(output) +{ + [Fact] + public async Task HandshakeCompletionGatePreventsUnauthenticatedPayloads() + { + // Simulate severe thread pool exhaustion using ThreadPool.SetMinThreads(1, 1); + ThreadPool.GetMinThreads(out int workerThreads, out int completionPortThreads); + ThreadPool.SetMinThreads(1, 1); + try + { + var options = new ConfigurationOptions + { + EndPoints = { { TestConfig.Current.MasterServer, TestConfig.Current.MasterPort } }, + Password = TestConfig.Current.MasterPassword, + AbortOnConnectFail = false, + AllowAdmin = true + }; + + await using var conn = await ConnectionMultiplexer.ConnectAsync(options); + var db = conn.GetDatabase(); + var server = conn.GetServer(TestConfig.Current.MasterServer, TestConfig.Current.MasterPort); + + // Trigger an asynchronous connection reset loop + var resetLoop = Task.Run(async () => + { + for (int i = 0; i < 5; i++) + { + server.SimulateConnectionFailure(SimulatedFailureType.AuthenticationFailure); + await Task.Delay(10); + } + }); + + // Concurrently launch 100 parallel Tasks trying to dispatch rapid PingAsync() operations. + int taskCount = 100; + var tasks = new Task[taskCount]; + for (int i = 0; i < taskCount; i++) + { + tasks[i] = Task.Run(async () => + { + for (int j = 0; j < 50; j++) + { + try + { + await db.PingAsync().ConfigureAwait(false); + } + catch (Exception ex) + { + // Assert that no commands throw a NOAUTH or protocol corruption exception + Assert.DoesNotContain("NOAUTH", ex.Message); + Assert.DoesNotContain("protocol", ex.Message.ToLowerInvariant()); + } + } + }); + } + + await Task.WhenAll(tasks); + await resetLoop; + } + finally + { + ThreadPool.SetMinThreads(workerThreads, completionPortThreads); + } + } +} From 0a71a91ab23def74b0bd2a4acfc341463e70a99c Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 24 Jun 2026 12:49:53 +0100 Subject: [PATCH 2/5] Remove unused Xunit.Abstractions using directive --- tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs index 2ab0a7306..3f8442f58 100644 --- a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs +++ b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs @@ -2,7 +2,6 @@ using System.Threading; using System.Threading.Tasks; using Xunit; -using Xunit.Abstractions; namespace StackExchange.Redis.Tests; From 851178629ebee0d5ea2679b3c821e8b08130b7ec Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 24 Jun 2026 13:14:55 +0100 Subject: [PATCH 3/5] Update ConnectionRaceTests.cs --- tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs index 3f8442f58..9895587f2 100644 --- a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs +++ b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs @@ -10,6 +10,9 @@ public class ConnectionRaceTests(ITestOutputHelper output) : TestBase(output) [Fact] public async Task HandshakeCompletionGatePreventsUnauthenticatedPayloads() { + #if RELEASE + Assert.Skip("Ignoring in CI, due to threading"); + #endif // Simulate severe thread pool exhaustion using ThreadPool.SetMinThreads(1, 1); ThreadPool.GetMinThreads(out int workerThreads, out int completionPortThreads); ThreadPool.SetMinThreads(1, 1); @@ -17,15 +20,15 @@ public async Task HandshakeCompletionGatePreventsUnauthenticatedPayloads() { var options = new ConfigurationOptions { - EndPoints = { { TestConfig.Current.MasterServer, TestConfig.Current.MasterPort } }, - Password = TestConfig.Current.MasterPassword, + EndPoints = { { TestConfig.Current.PrimaryServer, TestConfig.Current.PrimaryPort } }, + Password = TestConfig.Current.PrimaryPassword, AbortOnConnectFail = false, AllowAdmin = true }; await using var conn = await ConnectionMultiplexer.ConnectAsync(options); var db = conn.GetDatabase(); - var server = conn.GetServer(TestConfig.Current.MasterServer, TestConfig.Current.MasterPort); + var server = conn.GetServer(TestConfig.Current.PrimaryServer, TestConfig.Current.PrimaryPort); // Trigger an asynchronous connection reset loop var resetLoop = Task.Run(async () => From 8b9f829b0f12b0f3285c12dd3211458daf768d43 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 24 Jun 2026 14:13:08 +0100 Subject: [PATCH 4/5] Update ConnectionRaceTests.cs --- tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs index 9895587f2..0b9af7f9b 100644 --- a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs +++ b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs @@ -21,7 +21,6 @@ public async Task HandshakeCompletionGatePreventsUnauthenticatedPayloads() var options = new ConfigurationOptions { EndPoints = { { TestConfig.Current.PrimaryServer, TestConfig.Current.PrimaryPort } }, - Password = TestConfig.Current.PrimaryPassword, AbortOnConnectFail = false, AllowAdmin = true }; @@ -35,7 +34,7 @@ public async Task HandshakeCompletionGatePreventsUnauthenticatedPayloads() { for (int i = 0; i < 5; i++) { - server.SimulateConnectionFailure(SimulatedFailureType.AuthenticationFailure); + server.SimulateConnectionFailure(SimulatedFailureType.All); await Task.Delay(10); } }); From b53c407d6b3360d6e483b4d5b0560a3721bf44fb Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 24 Jun 2026 14:22:49 +0100 Subject: [PATCH 5/5] Fix formatting in ConnectionRaceTests.cs --- tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs index 0b9af7f9b..5579e9a15 100644 --- a/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs +++ b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs @@ -22,7 +22,7 @@ public async Task HandshakeCompletionGatePreventsUnauthenticatedPayloads() { EndPoints = { { TestConfig.Current.PrimaryServer, TestConfig.Current.PrimaryPort } }, AbortOnConnectFail = false, - AllowAdmin = true + AllowAdmin = true, }; await using var conn = await ConnectionMultiplexer.ConnectAsync(options);