diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs
index 8d01306d0..ab65eb095 100644
--- a/src/StackExchange.Redis/ConnectionMultiplexer.cs
+++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs
@@ -530,6 +530,25 @@ private bool WaitAllIgnoreErrors(Task[] tasks)
return false;
}
+ ///
+ /// An endpoint's availability for the reconfigure connect wait, when the dedicated subscription
+ /// connection is part of connecting ():
+ /// 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.
+ ///
+ private static async Task AlsoAwaitingTheSubscriptionLeg(Task connected, Task subscriptionConnected)
+ {
+ try
+ {
+ await subscriptionConnected.ForAwait();
+ }
+ catch
+ {
+ }
+ return await connected.ForAwait();
+ }
+
private static async Task 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 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
diff --git a/src/StackExchange.Redis/Contexts/RespConnectionManager.cs b/src/StackExchange.Redis/Contexts/RespConnectionManager.cs
index e1c35659b..85d71bd48 100644
--- a/src/StackExchange.Redis/Contexts/RespConnectionManager.cs
+++ b/src/StackExchange.Redis/Contexts/RespConnectionManager.cs
@@ -1374,6 +1374,17 @@ await SubscribeToConfigurationChannelAsync(context, cancellationToken)
}
}
+ ///
+ /// The dedicated subscription connection's availability, for the reconfigure connect wait: completes
+ /// when the eagerly-dialled socket () has established.
+ /// 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.
+ ///
+ /// The endpoint whose subscription leg the connect is waiting on.
+ internal Task SubscriptionConnectedAsync(EndPoint endpoint)
+ => SubscriptionEndpoint(endpoint).ConnectNowAsync(CancellationToken.None);
+
///
/// 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.
diff --git a/tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs b/tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs
index e25bc1768..9f81e4abc 100644
--- a/tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs
+++ b/tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs
@@ -324,6 +324,42 @@ public async Task ConfigurationChannelOpensSubscriptionConnectionEagerly(RedisPr
Assert.Equal(1, subscription.SubscriptionCount); // the configuration channel
}
+ ///
+ /// The deterministic shape of '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 ConnectAsync 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.
+ ///
+ [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()
{
diff --git a/tests/StackExchange.Redis.Tests/InProcessTestServer.cs b/tests/StackExchange.Redis.Tests/InProcessTestServer.cs
index 726eb0bbb..4c6f9a841 100644
--- a/tests/StackExchange.Redis.Tests/InProcessTestServer.cs
+++ b/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;
+ /// As , but seeing the connection type the client
+ /// declared, so a test can slow or refuse one leg of an endpoint without touching the other.
+ protected virtual ValueTask OnAcceptClientAsync(EndPoint endpoint, ConnectionType connectionType)
+ => OnAcceptClientAsync(endpoint);
+
/*
private readonly RespServer _server;