From 687303f89c50ac213229af42a00c71f6359f67ab Mon Sep 17 00:00:00 2001 From: mgravell Date: Thu, 12 Nov 2026 17:00:10 +0000 Subject: [PATCH] Pub/sub: overlapping subscription restores cannot register on two servers (#3336) During a failover, OnFullyEstablished (per reconnecting connection) and ReconfigureAsync both restore subscriptions fire-and-forget; between sending (P)SUBSCRIBE and its acknowledgement a SingleNodeSubscription recorded no server, so a concurrent restore would subscribe again - often on a different node. The later ack then overwrote the recorded endpoint while the earlier registration stayed live on its server, delivering every message twice until that connection happened to drop. Two layers: - claim before send: a pending-claim token (CAS) makes competing restores back off while an attempt is in flight; the subscribe message is sent without fire-and-forget (callers that asked for F&F just don't await it) so every completion - ack, error reply, connection failure, timeout - observably releases the claim, keeping retry liveness. - reconcile at ack: TryAdoptEndpoint rejects an ack when the subscription is already confirmed on a different connected server, and the processor sends a fire-and-forget unsubscribe pinned to the stray node - mopping up anything the claim cannot prevent (e.g. a timed-out attempt whose ack arrives late). Regression tests run the reporter's overlap scenario (held first ack, second restore must not subscribe elsewhere) plus the stray-ack cleanup, per protocol, on the in-process test server. --- src/StackExchange.Redis/ResultProcessor.cs | 19 +- src/StackExchange.Redis/Subscription.cs | 172 ++++++++++++++++-- .../SubscriptionRestoreRaceTests.cs | 133 ++++++++++++++ 3 files changed, 312 insertions(+), 12 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/SubscriptionRestoreRaceTests.cs diff --git a/src/StackExchange.Redis/ResultProcessor.cs b/src/StackExchange.Redis/ResultProcessor.cs index 50ce1255f..b4e91c001 100644 --- a/src/StackExchange.Redis/ResultProcessor.cs +++ b/src/StackExchange.Redis/ResultProcessor.cs @@ -565,7 +565,24 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes case RedisCommand.SUBSCRIBE: case RedisCommand.SSUBSCRIBE: case RedisCommand.PSUBSCRIBE: - Subscription?.AddEndpoint(ep); + if (Subscription is { } sub && !sub.TryAdoptEndpoint(ep)) + { + // the subscription is already confirmed on a different live server: + // this ack is a stray from an overlapping restore (#3336); remove + // the server-side registration it created, or it would deliver + // duplicate messages indefinitely (fire-and-forget: we are on the + // read loop, so we must not block) + if (message is Message.CommandChannelBase channelMessage) + { + var unsubscribe = sub.GetSubscriptionMessage( + channelMessage.Channel, + ConnectionMultiplexer.SubscriptionAction.Unsubscribe, + CommandFlags.FireAndForget, + internalCall: true); + connection.BridgeCouldBeNull?.Multiplexer.ExecuteSyncImpl(unsubscribe, this, ep); + } + } + break; default: Subscription?.TryRemoveEndpoint(ep); diff --git a/src/StackExchange.Redis/Subscription.cs b/src/StackExchange.Redis/Subscription.cs index 767ab0a14..681117032 100644 --- a/src/StackExchange.Redis/Subscription.cs +++ b/src/StackExchange.Redis/Subscription.cs @@ -25,7 +25,10 @@ internal abstract class Subscription internal abstract bool IsConnectedAny(); internal abstract bool IsConnectedTo(EndPoint endpoint); - internal abstract void AddEndpoint(ServerEndPoint server); + // Called when a (P|S)SUBSCRIBE acknowledgement arrives; returns false when the ack is a + // stray (we are already confirmed on a different, connected server) and the caller should + // remove the server-side registration it represents. + internal abstract bool TryAdoptEndpoint(ServerEndPoint server); // conditional clear internal abstract bool TryRemoveEndpoint(ServerEndPoint expected); @@ -77,7 +80,8 @@ internal Message GetSubscriptionMessage( in RedisChannel channel, SubscriptionAction action, CommandFlags flags, - bool internalCall) + bool internalCall, + bool stripFireAndForget = false) { const RedisChannel.RedisChannelOptions OPTIONS_MASK = ~( RedisChannel.RedisChannelOptions.KeyRouted | RedisChannel.RedisChannelOptions.IgnoreChannelPrefix); @@ -108,7 +112,17 @@ internal Message GetSubscriptionMessage( }; // TODO: Consider flags here - we need to pass Fire and Forget, but don't want to intermingle Primary/Replica - var msg = Message.Create(-1, Flags | flags, command, channel); + var effectiveFlags = Flags | flags; + if (stripFireAndForget) + { + // a fire-and-forget message has no result box, so nothing observes its completion; + // the subscribe-claim logic needs that signal (ack, error reply, connection failure + // or timeout) to release a pending claim, so the message itself must not be + // fire-and-forget even when the caller is (the caller then simply doesn't await it) + effectiveFlags &= ~CommandFlags.FireAndForget; + } + + var msg = Message.Create(-1, effectiveFlags, command, channel); msg.SetForSubscriptionBridge(); if (internalCall) { @@ -206,7 +220,30 @@ internal override bool IsConnectedTo(EndPoint endpoint) return server is { IsSubscriberConnected: true } && server.EndPoint == endpoint; } - internal override void AddEndpoint(ServerEndPoint server) => _currentServer = server; + internal override bool TryAdoptEndpoint(ServerEndPoint server) + { + // an ack from the pending claim's server resolves that claim, whether or not we adopt it + var pending = Volatile.Read(ref _pending); + if (pending is not null && pending.Server == server) + { + Interlocked.CompareExchange(ref _pending, null, pending); + } + + var current = Volatile.Read(ref _currentServer); + while (true) + { + if (current is { IsSubscriberConnected: true } && current != server) + { + // already confirmed on a different live server; the incoming ack is a stray + // from an overlapping restore, and should be unsubscribed by the caller + return false; + } + + var seen = Interlocked.CompareExchange(ref _currentServer, server, current); + if (seen == current) return true; + current = seen; + } + } internal override bool TryRemoveEndpoint(ServerEndPoint expected) { @@ -253,6 +290,12 @@ internal override Task UnsubscribeFromServerAsync( } private ServerEndPoint? _currentServer; + + // an in-flight subscribe attempt; while one is pending, other restore attempts back off, + // so that overlapping restores (reconnect + reconfigure racing during a failover) cannot + // register the same subscription on two different servers (#3336) + private PendingClaim? _pending; + internal ServerEndPoint? GetCurrentServer() => Volatile.Read(ref _currentServer); internal override ServerEndPoint? GetAnyCurrentServer() => Volatile.Read(ref _currentServer); @@ -267,6 +310,12 @@ internal override void RemoveDisconnectedEndpoints() { _currentServer = null; } + + var pending = Volatile.Read(ref _pending); + if (pending is not null && !pending.Server.IsSubscriberConnected) + { + Interlocked.CompareExchange(ref _pending, null, pending); + } } internal override int EnsureSubscribedToServer( @@ -278,14 +327,77 @@ internal override int EnsureSubscribedToServer( RemoveIncorrectRouting(subscriber, in channel, flags, internalCall); if (IsConnectedAny()) return 0; + bool callerIsFireAndForget = ((Flags | flags) & CommandFlags.FireAndForget) != 0; + var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall, stripFireAndForget: true); + var selected = subscriber.multiplexer.SelectServer(message); + if (!TryClaim(selected, out var claim)) return 0; + // we're not appropriately connected, so blank it out for eligible reconnection _currentServer = null; - var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall); - var selected = subscriber.multiplexer.SelectServer(message); - _ = subscriber.ExecuteSync(message, Processor, selected); + if (callerIsFireAndForget) + { + try + { + // fire-and-observe: never block (restore paths run inside result processing), + // but watch the outcome so a failed attempt releases the claim for retry + ObserveClaim(subscriber.ExecuteAsync(message, Processor, selected), claim); + } + catch + { + Release(claim); + throw; + } + } + else + { + try + { + _ = subscriber.ExecuteSync(message, Processor, selected); + } + finally + { + // normally already resolved by the ack; backstop for failures + Release(claim); + } + } + return 1; } + private bool TryClaim(ServerEndPoint? selected, out PendingClaim? claim) + { + if (selected is null) + { + // nothing selectable right now; attempt unclaimed (ack-time reconciliation + // still protects against duplicates) + claim = null; + return true; + } + + claim = new PendingClaim(this, selected); + // if another subscribe attempt is already in flight, back off; its completion + // (ack or failure) will resolve the state, retried via the usual restore paths + return Interlocked.CompareExchange(ref _pending, claim, null) is null; + } + + private void Release(PendingClaim? claim) + { + if (claim is not null) + { + Interlocked.CompareExchange(ref _pending, null, claim); + } + } + + private static void ObserveClaim(Task task, PendingClaim? claim) + => task.ContinueWith( + static (t, state) => + { + _ = t.Exception; // observed; the retry policy is simply "the next restore attempt" + (state as PendingClaim)?.Release(); + }, + claim, + TaskContinuationOptions.ExecuteSynchronously); + private void RemoveIncorrectRouting(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall) { // only applies to cluster, when using key-routed channels (sharded, explicit key-routed, or @@ -313,13 +425,48 @@ internal override async Task EnsureSubscribedToServerAsync( RemoveIncorrectRouting(subscriber, in channel, flags, internalCall); if (IsConnectedAny()) return 0; + bool callerIsFireAndForget = ((Flags | flags) & CommandFlags.FireAndForget) != 0; + var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall, stripFireAndForget: true); + server ??= subscriber.multiplexer.SelectServer(message); + if (!TryClaim(server, out var claim)) return 0; + // we're not appropriately connected, so blank it out for eligible reconnection _currentServer = null; - var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall); - server ??= subscriber.multiplexer.SelectServer(message); - await subscriber.ExecuteAsync(message, Processor, server).ForAwait(); + if (callerIsFireAndForget) + { + try + { + // fire-and-observe; see EnsureSubscribedToServer + ObserveClaim(subscriber.ExecuteAsync(message, Processor, server), claim); + } + catch + { + Release(claim); + throw; + } + } + else + { + try + { + await subscriber.ExecuteAsync(message, Processor, server).ForAwait(); + } + finally + { + // normally already resolved by the ack; backstop for failures + Release(claim); + } + } + return 1; } + + private sealed class PendingClaim(SingleNodeSubscription owner, ServerEndPoint server) + { + public ServerEndPoint Server { get; } = server; + + public void Release() => owner.Release(this); + } } // used for keyspace subscriptions, which are routed to multiple nodes @@ -352,13 +499,16 @@ internal override bool IsConnectedTo(EndPoint endpoint) => _servers.TryGetValue(endpoint, out var server) && server.IsSubscriberConnected; - internal override void AddEndpoint(ServerEndPoint server) + internal override bool TryAdoptEndpoint(ServerEndPoint server) { + // multi-node subscriptions want every eligible node, so an ack is never a stray var ep = server.EndPoint; if (!_servers.TryAdd(ep, server)) { _servers[ep] = server; } + + return true; } internal override bool TryRemoveEndpoint(ServerEndPoint expected) diff --git a/tests/StackExchange.Redis.Tests/SubscriptionRestoreRaceTests.cs b/tests/StackExchange.Redis.Tests/SubscriptionRestoreRaceTests.cs new file mode 100644 index 000000000..a2093bd3e --- /dev/null +++ b/tests/StackExchange.Redis.Tests/SubscriptionRestoreRaceTests.cs @@ -0,0 +1,133 @@ +using System; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis.Server; +using Xunit; + +namespace StackExchange.Redis.Tests; + +// Regression tests for https://github.com/seredis/StackExchange.Redis/issues/3336 - during a +// failover, overlapping subscription-restore paths could each (P)SUBSCRIBE on a different node +// before the first acknowledgement recorded a server, leaving a stray server-side registration +// that delivered duplicate pub/sub messages indefinitely. +[RunPerProtocol] +public class SubscriptionRestoreRaceTests(ITestOutputHelper output) +{ + [Theory] + [InlineData(false, false)] + [InlineData(true, false)] + [InlineData(false, true)] + [InlineData(true, true)] + public async Task OverlappingRestoresCannotSubscribeOnTwoServers(bool pattern, bool fireAndForget) + { + using var server = new PausedSubscribeServer(output); + await using var connection = await server.ConnectAsync(); + var subscriber = connection.DefaultSubscriber; + var channel = pattern ? RedisChannel.Pattern("restoration.*") : RedisChannel.Literal("restoration.channel"); + var subscription = new ConnectionMultiplexer.SingleNodeSubscription(fireAndForget ? CommandFlags.FireAndForget : CommandFlags.None); + var firstServer = connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured); + var secondServer = connection.GetServerEndPoint(server.SecondEndPoint, ServerProvenance.Configured); + var first = subscription.EnsureSubscribedToServerAsync(subscriber, channel, + fireAndForget ? CommandFlags.FireAndForget : CommandFlags.None, true, firstServer); + try + { + await server.FirstSubscribe.Task.WithTimeout(5000); + + // while the first subscribe is unacknowledged, a second restore must back off + var second = subscription.EnsureSubscribedToServerAsync(subscriber, channel, CommandFlags.None, true, secondServer); + var ping = Message.Create(-1, CommandFlags.None, RedisCommand.PING); + ping.SetForSubscriptionBridge(); + await connection.ExecuteAsyncImpl(ping, ResultProcessor.ResponseTimer, null, secondServer).WithTimeout(5000); + Assert.Equal(0, Volatile.Read(ref server.SecondSubscribeCount)); + + server.Release.Set(); + await Task.WhenAll(first, second).WithTimeout(5000); + await WaitForCurrentServerAsync(subscription); + Assert.Equal(1, Volatile.Read(ref server.FirstSubscribeCount)); + Assert.Equal(0, Volatile.Read(ref server.SecondSubscribeCount)); + Assert.Same(firstServer, subscription.GetAnyCurrentServer()); + } + finally + { + server.Release.Set(); + try { await first.WithTimeout(5000); } catch { } + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task StrayAcknowledgementIsUnsubscribed(bool pattern) + { + using var server = new PausedSubscribeServer(output); + await using var connection = await server.ConnectAsync(); + var subscriber = connection.DefaultSubscriber; + var channel = pattern ? RedisChannel.Pattern("restoration.*") : RedisChannel.Literal("restoration.channel"); + var subscription = new ConnectionMultiplexer.SingleNodeSubscription(CommandFlags.None); + var firstServer = connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured); + var secondServer = connection.GetServerEndPoint(server.SecondEndPoint, ServerProvenance.Configured); + + // confirm the subscription on the first server + server.Release.Set(); + await subscription.EnsureSubscribedToServerAsync(subscriber, channel, CommandFlags.None, true, firstServer).WithTimeout(5000); + await WaitForCurrentServerAsync(subscription); + Assert.Same(firstServer, subscription.GetAnyCurrentServer()); + + // simulate an overlapping restore whose ack arrives late: a subscribe pinned to the second + // server, bypassing the claim; the ack must be rejected and the registration removed + var stray = subscription.GetSubscriptionMessage(channel, ConnectionMultiplexer.SubscriptionAction.Subscribe, CommandFlags.None, true); + await connection.ExecuteAsyncImpl(stray, subscription.Processor, null, secondServer).WithTimeout(5000); + + await server.SecondUnsubscribe.Task.WithTimeout(5000); + Assert.Same(firstServer, subscription.GetAnyCurrentServer()); + Assert.Equal(1, Volatile.Read(ref server.SecondSubscribeCount)); + } + + private static async Task WaitForCurrentServerAsync(ConnectionMultiplexer.SingleNodeSubscription subscription) + { + // fire-and-forget variants return before the ack; allow it to land + for (int i = 0; subscription.GetAnyCurrentServer() is null && i < 50; i++) + { + await Task.Delay(100); + } + } + + private sealed class PausedSubscribeServer : InProcessTestServer + { + public EndPoint SecondEndPoint { get; } + public readonly ManualResetEventSlim Release = new(); + public readonly TaskCompletionSource FirstSubscribe = new(TaskCreationOptions.RunContinuationsAsynchronously); + public int FirstSubscribeCount, SecondSubscribeCount; + public readonly TaskCompletionSource SecondUnsubscribe = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public PausedSubscribeServer(ITestOutputHelper output) : base(output) => SecondEndPoint = AddEmptyNode(); + + public override TypedRedisValue Execute(RedisClient client, in RedisRequest request) + { + if ((request.IsString(0, "SUBSCRIBE"u8) || request.IsString(0, "PSUBSCRIBE"u8)) + && request.GetString(1).StartsWith("restoration.", StringComparison.Ordinal)) + { + TryGetNode(DefaultEndPoint, out var firstNode); + if (ReferenceEquals(client.Node, firstNode)) + { + Interlocked.Increment(ref FirstSubscribeCount); + FirstSubscribe.TrySetResult(true); + if (!Release.Wait(TimeSpan.FromSeconds(10))) throw new TimeoutException("Test subscription barrier timed out."); + } + else + { + Interlocked.Increment(ref SecondSubscribeCount); + } + } + var result = base.Execute(client, in request); + if ((request.IsString(0, "UNSUBSCRIBE"u8) || request.IsString(0, "PUNSUBSCRIBE"u8)) + && request.GetString(1).StartsWith("restoration.", StringComparison.Ordinal)) + { + TryGetNode(DefaultEndPoint, out var firstNode); + if (!ReferenceEquals(client.Node, firstNode)) SecondUnsubscribe.TrySetResult(true); + } + return result; + } + } +}