fix(bridge): enforce TaskCompletionSource handshake gate to prevent connection race

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<bool> _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<WriteResult> 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<WriteResult> WriteMessageTakingWriteLockAsync(PhysicalConnect
             }
         }
 
+        private async ValueTask<WriteResult> WriteMessageTakingWriteLockAsync_Gate(PhysicalConnection physical, Message message, bool bypassBacklog)
+        {
+            await _handshakeCompletion.Task.ConfigureAwait(false);
+            return await WriteMessageTakingWriteLockAsync(physical, message, bypassBacklog).ConfigureAwait(false);
+        }
+
         private async ValueTask<WriteResult> WriteMessageTakingWriteLockAsync_Awaited(
             ValueTask<bool> 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);

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);
+        }
+    }
+}