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..5579e9a15 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ConnectionRaceTests.cs @@ -0,0 +1,73 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +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); + try + { + var options = new ConfigurationOptions + { + EndPoints = { { TestConfig.Current.PrimaryServer, TestConfig.Current.PrimaryPort } }, + AbortOnConnectFail = false, + AllowAdmin = true, + }; + + await using var conn = await ConnectionMultiplexer.ConnectAsync(options); + var db = conn.GetDatabase(); + var server = conn.GetServer(TestConfig.Current.PrimaryServer, TestConfig.Current.PrimaryPort); + + // Trigger an asynchronous connection reset loop + var resetLoop = Task.Run(async () => + { + for (int i = 0; i < 5; i++) + { + server.SimulateConnectionFailure(SimulatedFailureType.All); + 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); + } + } +}