Add NeedsReconnect flag to defer reconnection to reader loop for MOVED-to-same-endpoint

Author: barshaul

src/StackExchange.Redis/PhysicalBridge.cs

@@ -48,6 +48,7 @@ internal sealed class PhysicalBridge : IDisposable
         private int failConnectCount = 0;
         private volatile bool isDisposed;
         private volatile bool shouldResetConnectionRetryCount;
+        private bool _needsReconnect;
         private long nonPreferredEndpointCount;
 
         // private volatile int missedHeartbeats;
@@ -131,6 +132,16 @@ public enum State : byte
         private RedisProtocol _protocol; // note starts at zero, not RESP2
         internal void SetProtocol(RedisProtocol protocol) => _protocol = protocol;
 
+        /// <summary>
+        /// Indicates whether the bridge needs to reconnect.
+        /// </summary>
+        internal bool NeedsReconnect => Volatile.Read(ref _needsReconnect);
+
+        /// <summary>
+        /// Marks that the bridge needs to reconnect.
+        /// </summary>
+        internal void MarkNeedsReconnect() => Volatile.Write(ref _needsReconnect, true);
+
         public void Dispose()
         {
             isDisposed = true;
@@ -210,7 +221,7 @@ private WriteResult FailDueToNoConnection(Message message)
         public WriteResult TryWriteSync(Message message, bool isReplica)
         {
             if (isDisposed) throw new ObjectDisposedException(Name);
-            if (!IsConnected) return QueueOrFailMessage(message);
+            if (!IsConnected || NeedsReconnect) return QueueOrFailMessage(message);
 
             var physical = this.physical;
             if (physical == null)
@@ -234,7 +245,7 @@ public WriteResult TryWriteSync(Message message, bool isReplica)
         public ValueTask<WriteResult> TryWriteAsync(Message message, bool isReplica, bool bypassBacklog = false)
         {
             if (isDisposed) throw new ObjectDisposedException(Name);
-            if (!IsConnected && !bypassBacklog) return new ValueTask<WriteResult>(QueueOrFailMessage(message));
+            if ((!IsConnected || NeedsReconnect) && !bypassBacklog) return new ValueTask<WriteResult>(QueueOrFailMessage(message));
 
             var physical = this.physical;
             if (physical == null)
@@ -1458,6 +1469,8 @@ private bool ChangeState(State oldState, State newState)
                         Multiplexer.Trace("Connecting...", Name);
                         if (ChangeState(State.Disconnected, State.Connecting))
                         {
+                            // Clear the reconnect flag as we're starting a new connection
+                            Volatile.Write(ref _needsReconnect, false);
                             Interlocked.Increment(ref socketCount);
                             Interlocked.Exchange(ref connectStartTicks, Environment.TickCount);
                             // separate creation and connection for case when connection completes synchronously

src/StackExchange.Redis/PhysicalConnection.cs

@@ -2091,9 +2091,9 @@ private async Task ReadFromPipe()
                     Trace($"Processed {handled} messages");
                     input.AdvanceTo(buffer.Start, buffer.End);
 
-                    if (handled == 0 && readResult.IsCompleted)
+                    if ((handled == 0 && readResult.IsCompleted) || BridgeCouldBeNull?.NeedsReconnect == true)
                     {
-                        break; // no more data, or trailing incomplete messages
+                        break; // no more data, trailing incomplete messages, or reconnection required
                     }
                 }
                 Trace("EOF");

src/StackExchange.Redis/ResultProcessor.cs

@@ -267,7 +267,8 @@ public virtual bool SetResult(PhysicalConnection connection, Message message, in
                             // This occurs when Redis/Valkey servers are behind DNS records, load balancers, or proxies.
                             // The MOVED error signals that the client should reconnect to allow the DNS/proxy/load balancer
                             // to route the connection to a different underlying server host, then retry the command.
-                            bridge?.TryConnect(null)?.Dispose();
+                            // Mark the bridge to reconnect - reader loop will handle disconnection and reconnection.
+                            bridge?.MarkNeedsReconnect();
                         }
                         if (bridge is null)
                         {

tests/StackExchange.Redis.Tests/MovedToSameEndpointTests.cs

@@ -77,6 +77,11 @@ public async Task MovedToSameEndpoint_TriggersReconnectAndRetry_CommandSucceeds(
             };
 
             await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
+            // Ping the server to ensure it's responsive
+            var server = conn.GetServer(listenEndpoint);
+            await server.PingAsync();
+            // Verify server is detected as cluster mode
+            Assert.Equal(ServerType.Cluster, server.ServerType);
             var db = conn.GetDatabase();
 
             // Record baseline counters after initial connection