v4: the connect wait covers the eagerly-dialled subscription connection

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -530,6 +530,25 @@ private bool WaitAllIgnoreErrors(Task[] tasks)
             return false;
         }
 
+        /// <summary>
+        /// An endpoint's availability for the reconfigure connect wait, when the dedicated subscription
+        /// connection is part of connecting (<see cref="ServerEndPoint.RequiresEagerSubscriptionConnection"/>):
+        /// complete only when both legs are. The subscription leg is best effort, mirroring the wait that
+        /// consumes this - its failure must not fault the endpoint's availability, because the interactive
+        /// leg decides selectability and a refused subscription dial is redialled on demand.
+        /// </summary>
+        private static async Task<string> AlsoAwaitingTheSubscriptionLeg(Task<string> connected, Task subscriptionConnected)
+        {
+            try
+            {
+                await subscriptionConnected.ForAwait();
+            }
+            catch
+            {
+            }
+            return await connected.ForAwait();
+        }
+
         private static async Task<bool> WaitAllIgnoreErrorsAsync(string name, Task[] tasks, int timeoutMilliseconds, ILogger? log, [CallerMemberName] string? caller = null, [CallerLineNumber] int callerLineNumber = 0)
         {
             _ = tasks ?? throw new ArgumentNullException(nameof(tasks));
@@ -1974,7 +1993,21 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, ILog
                             // This awaits either the endpoint's initial connection, or a tracer if we're already connected
                             // (which is the reconfigure case, except second iteration which is only for newly discovered cluster members).
                             var isFirstIteration = iter == 0;
-                            available[i] = server.OnConnectedAsync(log, sendTracerIfConnected: isFirstIteration, autoConfigureIfConnected: reconfigureAll);
+                            var connected = server.OnConnectedAsync(log, sendTracerIfConnected: isFirstIteration, autoConfigureIfConnected: reconfigureAll);
+
+                            // ...and the subscription leg, when it is part of connecting rather than on-demand:
+                            // ActivateServer above dialled it for the library's own configuration channel (or for
+                            // subscriptions that must be re-placed), but that dial is fire-and-forget, so released
+                            // on the interactive leg alone this wait lets ConnectAsync return a multiplexer whose
+                            // own broadcast channel is still dialling - which on a loaded machine it still is
+                            // (ConfigurationChannelOpensSubscriptionConnectionEagerly's CI failures). The v3 core
+                            // made its connect monitors wait on exactly this (#3310); here the monitors keep their
+                            // single-leg meaning - reconnect consumers want the interactive leg, and gating them
+                            // re-times subscription re-placement against one-shot publishes (Resp3DowngradeTests) -
+                            // and only this connect wait broadens.
+                            available[i] = server.RequiresEagerSubscriptionConnection
+                                ? AlsoAwaitingTheSubscriptionLeg(connected, Connections.SubscriptionConnectedAsync(server.EndPoint))
+                                : connected;
                         }
 
                         // One topology read per pass, by the core that owns the connections - replacing the

src/StackExchange.Redis/Contexts/RespConnectionManager.cs

@@ -1374,6 +1374,17 @@ await SubscribeToConfigurationChannelAsync(context, cancellationToken)
             }
         }
 
+        /// <summary>
+        /// The dedicated subscription connection's availability, for the reconfigure connect wait: completes
+        /// when the eagerly-dialled socket (<see cref="DialSubscriptionSocketIfRequired"/>) has established.
+        /// <see cref="RespEndpointExecutor.ConnectNowAsync"/> short-circuits when it already has, joins the
+        /// dial in flight when it has not, and declines onto a completed task under backoff - so the wait is
+        /// bounded by the caller's budget, never hung on a leg that is not coming.
+        /// </summary>
+        /// <param name="endpoint">The endpoint whose subscription leg the connect is waiting on.</param>
+        internal Task SubscriptionConnectedAsync(EndPoint endpoint)
+            => SubscriptionEndpoint(endpoint).ConnectNowAsync(CancellationToken.None);
+
         /// <summary>
         /// Dial a subscription socket when activation already knows it will be needed: for the library's own
         /// configuration channel, or for subscriptions that already exist and may need to move to this endpoint.

tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs

@@ -324,6 +324,42 @@ public async Task ConfigurationChannelOpensSubscriptionConnectionEagerly(RedisPr
         Assert.Equal(1, subscription.SubscriptionCount); // the configuration channel
     }
 
+    /// <summary>
+    /// The deterministic shape of <see cref="ConfigurationChannelOpensSubscriptionConnectionEagerly"/>'s CI
+    /// failures: the eager subscription connection is dialled fire-and-forget, so on a loaded machine the
+    /// connect wait released on the interactive leg alone and <c>ConnectAsync</c> returned while the
+    /// library's own broadcast channel was still dialling. A server slow to accept that one leg must delay
+    /// the connect, not escape it.
+    /// </summary>
+    [Theory]
+    [InlineData(RedisProtocol.Resp2)]
+    [InlineData(RedisProtocol.Resp3)]
+    public async Task ConfigurationChannelConnectWaitsForTheSubscriptionConnection(RedisProtocol protocol)
+    {
+        using var serverObj = new SlowSubscriptionAcceptServer(Output);
+        var config = serverObj.GetClientConfig();
+        config.Protocol = protocol;
+        config.ConfigurationChannel = nameof(ConfigurationChannelConnectWaitsForTheSubscriptionConnection);
+
+        await using var conn = await ConnectionMultiplexer.ConnectAsync(config, Writer);
+
+        var mux = (IInternalConnectionMultiplexer)conn;
+        var endpoint = conn.GetEndPoints().Single();
+        Assert.NotNull(mux.GetConnectionId(endpoint, ConnectionType.Interactive));
+        Assert.NotNull(mux.GetConnectionId(endpoint, ConnectionType.Subscription));
+        Assert.Equal(2, serverObj.ClientCount);
+    }
+
+    private sealed class SlowSubscriptionAcceptServer(ITestOutputHelper log) : InProcessTestServer(log)
+    {
+        protected override async ValueTask OnAcceptClientAsync(EndPoint endpoint, ConnectionType connectionType)
+        {
+            // long enough that a connect released on the interactive leg alone asserts before this accept
+            // completes, and comfortably inside the connect timeout for the wait that covers it
+            if (connectionType == ConnectionType.Subscription) await Task.Delay(1000);
+        }
+    }
+
     [Fact]
     public async Task VanillaResp2ConnectsWithSeparatePubSubConnection()
     {

tests/StackExchange.Redis.Tests/InProcessTestServer.cs

@@ -398,7 +398,7 @@ private sealed class InProcTunnel(
         {
             if (server.TryGetNode(endpoint, out var node))
             {
-                await server.OnAcceptClientAsync(endpoint);
+                await server.OnAcceptClientAsync(endpoint, connectionType);
                 server._log?.WriteLine(
                     $"[{endpoint}] accepting {connectionType} mapped to {server.ServerType} node {node} via {(server.UseSsl ? "TLS" : "plaintext")}");
                 var clientToServer = new Pipe(pipeOptions ?? PipeOptions.Default);
@@ -480,6 +480,11 @@ private sealed class StreamDuplexPipe(Stream stream) : IDuplexPipe
 
     protected virtual ValueTask OnAcceptClientAsync(EndPoint endpoint) => default;
 
+    /// <summary>As <see cref="OnAcceptClientAsync(EndPoint)"/>, but seeing the connection type the client
+    /// declared, so a test can slow or refuse one leg of an endpoint without touching the other.</summary>
+    protected virtual ValueTask OnAcceptClientAsync(EndPoint endpoint, ConnectionType connectionType)
+        => OnAcceptClientAsync(endpoint);
+
     /*
 
     private readonly RespServer _server;