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; + } + } +}