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;