PhysicalBridge: Fix orphaning cases of PhysicalConnection where it never finishes (#2338)

What happens here is that `PhysicalConnection` attempts to connect but never hears back, so it's stuck in `State.ConnectedEstablishing` state, which is handled in the heartbeat code. In the situation where the "last write seconds ago" passes in the heartbeat and we haven't heard back, we fire an `PhysicalBridge.OnDisconnected()` which clears out `physical` and orphans the socket listening forever. This now properly disposes of that `PhysicalConnection` mirroring like we do in `State.Connecting` which will properly fire `OnDisconnected()` and clean up the orphan socket.

The situation manifests where we establish a TCP connection, but not a Redis connection. All of the cases in the memory dump we're analyzing are some bytes sent and 0 received. Likely a Redis server issue, but the client is then handling it incorrectly and leaking.

I nuked the unused `RemovePhysical` method just to prevent further oops here.

Addresses a new case of #1458.

Author: NickCraver

docs/ReleaseNotes.md

@@ -8,7 +8,8 @@ Current package versions:
 
 ## Unreleased
 
-- Internal: revisit endpoint-snapshot implementation
+- Fix [#1458](https://github.com/StackExchange/StackExchange.Redis/issues/1458): Fixes a leak condition when a connection completes on the TCP phase but not the Redis handshake ([#2238 by NickCraver](https://github.com/StackExchange/StackExchange.Redis/pull/2238))
+- Internal: ServerSnapshot: Improve API and allow filtering with custom struct enumerator ([#2337 by mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/2337))
 
 
 ## 2.6.86

src/StackExchange.Redis/PhysicalBridge.cs

@@ -542,11 +542,16 @@ internal void OnHeartbeat(bool ifConnectedOnly)
                             // abort and reconnect
                             var snapshot = physical;
                             OnDisconnected(ConnectionFailureType.UnableToConnect, snapshot, out bool isCurrent, out State oldState);
-                            using (snapshot) { } // dispose etc
+                            snapshot?.Dispose(); // Cleanup the existing connection/socket if any, otherwise it will wait reading indefinitely
                             TryConnect(null);
                         }
                         break;
                     case (int)State.ConnectedEstablishing:
+                        // (Fall through) Happens when we successfully connected via TCP, but no Redis handshake completion yet.
+                        // This can happen brief (usual) or when the server never answers (rare). When we're in this state,
+                        // a socket is open and reader likely listening indefinitely for incoming data on an async background task.
+                        // We need to time that out and cleanup the PhysicalConnection if needed, otherwise that reader and socket will remain open
+                        // for the lifetime of the application due to being orphaned, yet still referenced by the active task doing the pipe read.
                     case (int)State.ConnectedEstablished:
                         var tmp = physical;
                         if (tmp != null)
@@ -576,6 +581,7 @@ internal void OnHeartbeat(bool ifConnectedOnly)
                                 else
                                 {
                                     OnDisconnected(ConnectionFailureType.SocketFailure, tmp, out bool ignore, out State oldState);
+                                    tmp.Dispose(); // Cleanup the existing connection/socket if any, otherwise it will wait reading indefinitely
                                 }
                             }
                             else if (writeEverySeconds <= 0 && tmp.IsIdle()
@@ -614,9 +620,6 @@ internal void OnHeartbeat(bool ifConnectedOnly)
             }
         }
 
-        internal void RemovePhysical(PhysicalConnection connection) =>
-            Interlocked.CompareExchange(ref physical, null, connection);
-
         [Conditional("VERBOSE")]
         internal void Trace(string message) => Multiplexer.Trace(message, ToString());