From 7ceeb7d4b896b0fd24b4d13545fc4b1405aaba66 Mon Sep 17 00:00:00 2001 From: mgravell Date: Sat, 31 Oct 2026 13:47:41 +0000 Subject: [PATCH] The connect wait covers the eagerly-dialled subscription connection A configuration channel makes the dedicated subscription connection part of connecting (#3310): ActivateServer dials it eagerly, but the dial is fire-and-forget by design, and the reconfigure connect wait covered the interactive leg alone - so on a loaded machine ConnectAsync returned while the library's own broadcast channel was still dialling. ConfigurationChannelOpensSubscriptionConnectionEagerly (merged with the 3.4.29 sync) fails exactly there on CI, on both OSes, while fast local machines win the race. The wait itself broadens: when RequiresEagerSubscriptionConnection, the endpoint's availability task also awaits the subscription executor's ConnectNowAsync (short-circuits when up, joins the dial in flight, declines onto a completed task under backoff, so the wait stays bounded by the pass's budget), with a subscription-leg failure deliberately not faulting the endpoint's availability - the interactive leg decides selectability, and a refused subscription dial redials on demand. Deliberately NOT by gating the connect monitors on the second leg, which is how the v3 core does it (OnFullyEstablished): those monitors complete on every reconnect, and gating them there re-times subscription re-placement against one-shot publishes - Resp3DowngradeTests' SubscribeQueuedBeforeDowngradeDoesNotPoisonInteractive fails deterministically with the PUBLISH answered :0, the re-SUBSCRIBE a beat behind it. The monitors keep their single-leg meaning; only the reconfigure wait, which is what ConnectAsync means by connected, grows the second leg. The new test pins the race deterministically: an in-process server that is slow to accept only the subscription leg (OnAcceptClientAsync now has an overload seeing the declared connection type) must delay ConnectAsync, not be escaped by it. Red on unfixed source with the CI failure's exact shape; green with the fix, both protocols. --- .../ConnectionMultiplexer.cs | 35 +++++++++++++++++- .../Contexts/RespConnectionManager.cs | 11 ++++++ .../DefaultOptionsTests.cs | 36 +++++++++++++++++++ .../InProcessTestServer.cs | 7 +++- 4 files changed, 87 insertions(+), 2 deletions(-) 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;