From 7de1db3b3607f1654fedfc47ba7adc5d5e83f6bb Mon Sep 17 00:00:00 2001 From: vkyow <218959617+vkyow@users.noreply.github.com> Date: Fri, 6 Nov 2026 12:22:58 +0800 Subject: [PATCH 1/3] Fix concurrent single-node Pub/Sub subscription restoration --- docs/PubSubOrder.md | 6 + .../PhysicalConnection.Read.cs | 4 +- src/StackExchange.Redis/RedisSubscriber.cs | 8 +- src/StackExchange.Redis/ResultProcessor.cs | 9 +- src/StackExchange.Redis/Subscription.cs | 321 ++++++++++++++---- .../SubscriptionRestorationTests.cs | 192 +++++++++++ 6 files changed, 461 insertions(+), 79 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs diff --git a/docs/PubSubOrder.md b/docs/PubSubOrder.md index 9a71049621..d0599c16d2 100644 --- a/docs/PubSubOrder.md +++ b/docs/PubSubOrder.md @@ -23,3 +23,9 @@ multiplexer.GetSubscriber().Subscribe("messages", (channel, message) => { Console.WriteLine((string)message); }); ``` +Reconnection +--- + +Single-node subscription restoration is coordinated until Redis acknowledges the subscription. Concurrent restoration triggers cannot establish independent routes for the same logical subscription, and late acknowledgements do not replace a newer confirmed connection. This applies to callback and queue subscriptions, including pattern channels; the processing-order choices above remain unchanged. + +Redis Pub/Sub does not replay messages missed during a disconnect or failover. Work already queued for a handler may still complete after unsubscribe; applications requiring durable delivery need a replay-capable messaging design. diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 5b572c4717..190f3f183f 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -694,13 +694,13 @@ private void OnMessage( { case RespPrefix.BulkString: case RespPrefix.SimpleString: - muxer.OnMessage(subscriptionChannel, messageChannel, reader.ReadRedisValue()); + muxer.OnMessage(subscriptionChannel, messageChannel, reader.ReadRedisValue(), this); break; case RespPrefix.Array: var iter = reader.AggregateChildren(); while (iter.MoveNext()) { - muxer.OnMessage(subscriptionChannel, messageChannel, iter.Value.ReadRedisValue()); + muxer.OnMessage(subscriptionChannel, messageChannel, iter.Value.ReadRedisValue(), this); } break; diff --git a/src/StackExchange.Redis/RedisSubscriber.cs b/src/StackExchange.Redis/RedisSubscriber.cs index d0741e5f5a..0becb4dc60 100644 --- a/src/StackExchange.Redis/RedisSubscriber.cs +++ b/src/StackExchange.Redis/RedisSubscriber.cs @@ -76,11 +76,15 @@ internal bool GetSubscriberCounts(in RedisChannel channel, out int handlers, out /// Handler that executes whenever a message comes in, this doles out messages to any registered handlers. /// internal void OnMessage(in RedisChannel subscription, in RedisChannel channel, in RedisValue payload) + => OnMessage(subscription, channel, payload, null); + + internal void OnMessage(in RedisChannel subscription, in RedisChannel channel, in RedisValue payload, PhysicalConnection? source) { ICompletable? completable = null; ChannelMessageQueue? queues = null; if (subscriptions.TryGetValue(subscription, out Subscription? sub)) { + if (source is not null && sub is SingleNodeSubscription single && !single.Accepts(source)) return; completable = sub.ForInvoke(channel, payload, out queues); } if (queues != null) @@ -263,7 +267,9 @@ internal void ResubscribeToServer(Subscription sub, in RedisChannel channel, Ser // we'll *try* for a simple resubscribe, following any -MOVED etc, but if that fails: fall back // to full reconfigure; importantly, note that we've already recorded the disconnect var message = sub.GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, CommandFlags.None, false); - _ = ExecuteAsync(message, sub.Processor, serverEndPoint).ContinueWith( + _ = (sub is SingleNodeSubscription + ? (Task)sub.EnsureSubscribedToServerAsync(this, channel, CommandFlags.None, false, serverEndPoint) + : ExecuteAsync(message, sub.Processor, serverEndPoint)).ContinueWith( t => multiplexer.ReconfigureIfNeeded(serverEndPoint.EndPoint, false, cause: cause), TaskContinuationOptions.OnlyOnFaulted); } diff --git a/src/StackExchange.Redis/ResultProcessor.cs b/src/StackExchange.Redis/ResultProcessor.cs index 7df8b83219..ec2fe82d48 100644 --- a/src/StackExchange.Redis/ResultProcessor.cs +++ b/src/StackExchange.Redis/ResultProcessor.cs @@ -537,7 +537,10 @@ protected override void WriteImpl(in MessageWriter writer) public sealed class TrackSubscriptionsProcessor : ResultProcessor { private ConnectionMultiplexer.Subscription? Subscription { get; } + private readonly Action? _onSubscribed; public TrackSubscriptionsProcessor(ConnectionMultiplexer.Subscription? sub) => Subscription = sub; + internal TrackSubscriptionsProcessor(ConnectionMultiplexer.Subscription sub, Action onSubscribed) + : this(sub) => _onSubscribed = onSubscribed; protected override bool SetResultCore(PhysicalConnection connection, Message message, ref RespReader reader) { @@ -555,8 +558,6 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes if (iter.Value.TryReadInt64(out long count)) { connection.SubscriptionCount = count; - SetResult(message, true); - var ep = connection.BridgeCouldBeNull?.ServerEndPoint; if (ep is not null) { @@ -565,13 +566,15 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes case RedisCommand.SUBSCRIBE: case RedisCommand.SSUBSCRIBE: case RedisCommand.PSUBSCRIBE: - Subscription?.AddEndpoint(ep); + if (_onSubscribed is not null) _onSubscribed(connection); + else Subscription?.AddEndpoint(ep); break; default: Subscription?.TryRemoveEndpoint(ep); break; } } + SetResult(message, true); return true; } } diff --git a/src/StackExchange.Redis/Subscription.cs b/src/StackExchange.Redis/Subscription.cs index 767ab0a143..1e70c3b7a8 100644 --- a/src/StackExchange.Redis/Subscription.cs +++ b/src/StackExchange.Redis/Subscription.cs @@ -77,7 +77,8 @@ internal Message GetSubscriptionMessage( in RedisChannel channel, SubscriptionAction action, CommandFlags flags, - bool internalCall) + bool internalCall, + bool trackReply = false) { const RedisChannel.RedisChannelOptions OPTIONS_MASK = ~( RedisChannel.RedisChannelOptions.KeyRouted | RedisChannel.RedisChannelOptions.IgnoreChannelPrefix); @@ -108,7 +109,9 @@ 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 (trackReply) effectiveFlags &= ~CommandFlags.FireAndForget; + var msg = Message.Create(-1, effectiveFlags, command, channel); msg.SetForSubscriptionBridge(); if (internalCall) { @@ -196,94 +199,284 @@ internal void GetSubscriberCounts(out int handlers, out int queues) // used for most subscriptions; routed to a single node internal sealed class SingleNodeSubscription(CommandFlags flags) : Subscription(flags) { - internal override bool IsConnectedAny() => _currentServer is { IsSubscriberConnected: true }; - - internal override int GetConnectionCount() => IsConnectedAny() ? 1 : 0; + private readonly SemaphoreSlim _operationLock = new(1, 1); + private readonly object _stateLock = new(); + private ServerEndPoint? _currentServer; + private PhysicalConnection? _currentConnection; + private RestoreOperation? _pendingRestore; + private System.Collections.Generic.List? _cleanup; + private bool _unsubscribed; - internal override bool IsConnectedTo(EndPoint endpoint) + internal override bool IsConnectedAny() { - var server = _currentServer; - return server is { IsSubscriberConnected: true } && server.EndPoint == endpoint; + var connection = Volatile.Read(ref _currentConnection); + return _currentServer is { IsSubscriberConnected: true } + && (connection is null || ReferenceEquals(connection.BridgeCouldBeNull?.Physical, connection)); } + internal override int GetConnectionCount() => IsConnectedAny() ? 1 : 0; + internal override bool IsConnectedTo(EndPoint endpoint) + => _currentServer is { IsSubscriberConnected: true } server && server.EndPoint == endpoint; + + internal bool Accepts(PhysicalConnection connection) + => !Volatile.Read(ref _unsubscribed) && ReferenceEquals(Volatile.Read(ref _currentConnection), connection); - internal override void AddEndpoint(ServerEndPoint server) => _currentServer = server; + internal override void AddEndpoint(ServerEndPoint server) { lock (_stateLock) _currentServer = server; } internal override bool TryRemoveEndpoint(ServerEndPoint expected) { - if (_currentServer == expected) + lock (_stateLock) { + if (_currentServer != expected) return false; _currentServer = null; + Volatile.Write(ref _currentConnection, null); return true; } + } - return false; + internal ServerEndPoint? GetCurrentServer() => Volatile.Read(ref _currentServer); + internal override ServerEndPoint? GetAnyCurrentServer() => GetCurrentServer(); + + internal override void RemoveDisconnectedEndpoints() + { + lock (_stateLock) + { + if (_currentServer is not null && !IsConnectedAny()) + { + _currentServer = null; + Volatile.Write(ref _currentConnection, null); + } + } } - internal override bool UnsubscribeFromServer( - RedisSubscriber subscriber, - in RedisChannel channel, - CommandFlags flags, - bool internalCall) + internal override int EnsureSubscribedToServer(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall) { - var server = _currentServer; - if (server is not null) + bool fireAndForget = (flags & CommandFlags.FireAndForget) != 0; + if (!Enter(subscriber, fireAndForget)) return 0; + RestoreOperation? operation = null; + bool transferred = false; + try { - var message = GetSubscriptionMessage(channel, SubscriptionAction.Unsubscribe, flags, internalCall); - return subscriber.multiplexer.ExecuteSyncImpl(message, Processor, server); + operation = BeginRestore(subscriber, channel, flags, internalCall, null); + if (operation is null) return 0; + if (fireAndForget) + { + transferred = true; + _ = CompleteRestoreAsync(operation).ObserveErrors(); + return 1; + } + subscriber.ExecuteSync(operation.Message, operation.Processor, operation.Server); + return 1; + } + finally + { + if (!transferred) EndOperation(operation); + } + } + + internal override async Task EnsureSubscribedToServerAsync(RedisSubscriber subscriber, RedisChannel channel, CommandFlags flags, bool internalCall, ServerEndPoint? server = null) + { + bool fireAndForget = (flags & CommandFlags.FireAndForget) != 0; + if (fireAndForget) + { + if (!Enter(subscriber, true)) return 0; + } + else if (!await _operationLock.WaitAsync(subscriber.multiplexer.AsyncTimeoutMilliseconds).ForAwait()) + { + throw WaitingTimeout(subscriber); + } + RestoreOperation? operation = null; + bool transferred = false; + try + { + operation = BeginRestore(subscriber, channel, flags, internalCall, server); + if (operation is null) return 0; + if (fireAndForget) + { + transferred = true; + _ = CompleteRestoreAsync(operation).ObserveErrors(); + return 1; + } + await subscriber.ExecuteAsync(operation.Message, operation.Processor, operation.Server).ForAwait(); + return 1; } + finally { if (!transferred) EndOperation(operation); } + } - return true; + private bool Enter(RedisSubscriber subscriber, bool fireAndForget) + { + if (_operationLock.Wait(fireAndForget ? 0 : subscriber.multiplexer.TimeoutMilliseconds)) return true; + if (fireAndForget) return false; + throw WaitingTimeout(subscriber); } - internal override Task UnsubscribeFromServerAsync( - RedisSubscriber subscriber, - RedisChannel channel, - CommandFlags flags, - object? asyncState, - bool internalCall) + private Exception WaitingTimeout(RedisSubscriber subscriber) + => ExceptionFactory.Timeout(subscriber.multiplexer, "Timeout waiting for an in-flight subscription operation.", null, _currentServer); + + private RestoreOperation? BeginRestore(RedisSubscriber subscriber, RedisChannel channel, CommandFlags flags, bool internalCall, ServerEndPoint? server) { - var server = _currentServer; - if (server is not null) + lock (_stateLock) { - var message = GetSubscriptionMessage(channel, SubscriptionAction.Unsubscribe, flags, internalCall); - return subscriber.multiplexer.ExecuteAsyncImpl(message, Processor, asyncState, server); + if (_unsubscribed) return null; + RemoveIncorrectRouting(subscriber, channel, flags, internalCall); + if (IsConnectedAny()) return null; + _currentServer = null; + Volatile.Write(ref _currentConnection, null); + // Keep an actual reply task even for fire-and-forget restoration: sending is not acknowledgement. + var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall, trackReply: true); + server ??= subscriber.multiplexer.SelectServer(message); + return _pendingRestore = new RestoreOperation(this, subscriber, channel, message, server); } + } - return CompletedTask.FromResult(true, asyncState); + private async Task CompleteRestoreAsync(RestoreOperation operation) + { + try { await operation.Subscriber.ExecuteAsync(operation.Message, operation.Processor, operation.Server).ForAwait(); } + finally { EndOperation(operation); } } - private ServerEndPoint? _currentServer; - internal ServerEndPoint? GetCurrentServer() => Volatile.Read(ref _currentServer); + private void EndOperation(RestoreOperation? operation) + { + try + { + lock (_stateLock) + { + if (ReferenceEquals(_pendingRestore, operation)) _pendingRestore = null; + DrainCleanup(); + } + } + finally { _operationLock.Release(); } + } + + private void OnSubscribed(RestoreOperation operation, PhysicalConnection connection) + { + var server = connection.BridgeCouldBeNull?.ServerEndPoint; + if (server is null) return; + lock (_stateLock) + { + if (ReferenceEquals(_pendingRestore, operation)) + { + _currentServer = server; + Volatile.Write(ref _currentConnection, connection); + return; + } + // A timed-out request can still be acknowledged. It must not replace a newer route. + QueueCleanup(new CleanupRequest(operation.Subscriber, operation.Channel, server)); + } + } - internal override ServerEndPoint? GetAnyCurrentServer() => Volatile.Read(ref _currentServer); + private void QueueCleanup(CleanupRequest request) + { + lock (_stateLock) + { + (_cleanup ??= new()).Add(request); + DrainCleanup(); + } + } - /// - /// Evaluates state and if we're not currently connected, clears the server reference. - /// - internal override void RemoveDisconnectedEndpoints() + private void DrainCleanup() { - var server = _currentServer; - if (server is { IsSubscriberConnected: false }) + // A redirected restore may not have its final endpoint yet. Reconcile after its acknowledgement. + if (_pendingRestore is not null || _cleanup is null) return; + var cleanup = _cleanup; + _cleanup = null; + foreach (var request in cleanup) { - _currentServer = null; + var muxer = request.Subscriber.multiplexer; + if (muxer.TryGetSubscription(request.Channel, out var active) && !ReferenceEquals(active, this)) + { + if (active is SingleNodeSubscription single) + { + single.QueueCleanup(request); + continue; + } + if (active.IsConnectedTo(request.Server.EndPoint)) continue; + } + if (!_unsubscribed && ReferenceEquals(_currentServer, request.Server)) continue; + try + { + var message = GetSubscriptionMessage(request.Channel, SubscriptionAction.Unsubscribe, CommandFlags.FireAndForget, true); + muxer.ExecuteAsyncImpl(message, ResultProcessor.TrackSubscriptions, null, request.Server).ObserveErrors(); + } + catch (Exception ex) { muxer.OnInternalError(ex); } } } - internal override int EnsureSubscribedToServer( - RedisSubscriber subscriber, - in RedisChannel channel, - CommandFlags flags, - bool internalCall) + internal override bool UnsubscribeFromServer(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall) + { + Volatile.Write(ref _unsubscribed, true); + if ((flags & CommandFlags.FireAndForget) != 0) + { + UnsubscribeCoreAsync(subscriber, channel, flags, null, internalCall).ObserveErrors(); + return true; + } + Enter(subscriber, false); + try + { + var server = _currentServer; + if (server is null) return true; + if (ForwardCleanup(subscriber, channel, server)) return true; + var message = GetSubscriptionMessage(channel, SubscriptionAction.Unsubscribe, flags, internalCall, trackReply: true); + return subscriber.multiplexer.ExecuteSyncImpl(message, Processor, server); + } + finally { EndOperation(null); } + } + + internal override Task UnsubscribeFromServerAsync(RedisSubscriber subscriber, RedisChannel channel, CommandFlags flags, object? asyncState, bool internalCall) + { + Volatile.Write(ref _unsubscribed, true); + var task = UnsubscribeCoreAsync(subscriber, channel, flags, asyncState, internalCall); + if ((flags & CommandFlags.FireAndForget) == 0) return task; + task.ObserveErrors(); + return CompletedTask.FromResult(true, asyncState); + } + + private async Task UnsubscribeCoreAsync(RedisSubscriber subscriber, RedisChannel channel, CommandFlags flags, object? asyncState, bool internalCall) + { + if (!await _operationLock.WaitAsync(subscriber.multiplexer.AsyncTimeoutMilliseconds).ForAwait()) throw WaitingTimeout(subscriber); + try + { + var server = _currentServer; + if (server is null) return true; + if (ForwardCleanup(subscriber, channel, server)) return true; + var message = GetSubscriptionMessage(channel, SubscriptionAction.Unsubscribe, flags, internalCall, trackReply: true); + return await subscriber.multiplexer.ExecuteAsyncImpl(message, Processor, asyncState, server).ForAwait(); + } + finally { EndOperation(null); } + } + + private bool ForwardCleanup(RedisSubscriber subscriber, RedisChannel channel, ServerEndPoint server) + { + if (subscriber.multiplexer.TryGetSubscription(channel, out var active) && !ReferenceEquals(active, this)) + { + if (active is SingleNodeSubscription single) single.QueueCleanup(new CleanupRequest(subscriber, channel, server)); + else if (!active.IsConnectedTo(server.EndPoint)) QueueCleanup(new CleanupRequest(subscriber, channel, server)); + return true; + } + return false; + } + + private sealed class RestoreOperation { - RemoveIncorrectRouting(subscriber, in channel, flags, internalCall); - if (IsConnectedAny()) return 0; + public RestoreOperation(SingleNodeSubscription owner, RedisSubscriber subscriber, RedisChannel channel, Message message, ServerEndPoint? server) + { + Subscriber = subscriber; + Channel = channel; + Message = message; + Server = server; + Processor = new ResultProcessor.TrackSubscriptionsProcessor(owner, connection => owner.OnSubscribed(this, connection)); + } + public RedisSubscriber Subscriber { get; } + public RedisChannel Channel { get; } + public Message Message { get; } + public ServerEndPoint? Server { get; } + public ResultProcessor.TrackSubscriptionsProcessor Processor { get; } + } - // 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); - return 1; + private sealed class CleanupRequest(RedisSubscriber subscriber, RedisChannel channel, ServerEndPoint server) + { + public RedisSubscriber Subscriber { get; } = subscriber; + public RedisChannel Channel { get; } = channel; + public ServerEndPoint Server { get; } = server; } private void RemoveIncorrectRouting(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall) @@ -302,24 +495,6 @@ private void RemoveIncorrectRouting(RedisSubscriber subscriber, in RedisChannel } } } - - internal override async Task EnsureSubscribedToServerAsync( - RedisSubscriber subscriber, - RedisChannel channel, - CommandFlags flags, - bool internalCall, - ServerEndPoint? server = null) - { - RemoveIncorrectRouting(subscriber, in channel, flags, internalCall); - if (IsConnectedAny()) 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(); - return 1; - } } // used for keyspace subscriptions, which are routed to multiple nodes diff --git a/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs b/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs new file mode 100644 index 0000000000..cb7a0a9a52 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs @@ -0,0 +1,192 @@ +using System; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis.Server; +using Xunit; + +namespace StackExchange.Redis.Tests; + +[RunPerProtocol] +public class SubscriptionRestorationTests(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); + 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); + 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 SynchronousFireAndForgetRestoreCoalescesWithPendingAcknowledgement(bool pattern) + { + using var server = new PausedSubscribeServer(output); + await using var connection = await server.ConnectAsync(); + 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 first = subscription.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true, firstServer); + try + { + await server.FirstSubscribe.Task.WithTimeout(5000); + Assert.Equal(0, subscription.EnsureSubscribedToServer(connection.DefaultSubscriber, channel, CommandFlags.FireAndForget, true)); + server.Release.Set(); + await first.WithTimeout(5000); + Assert.Equal(1, server.FirstSubscribeCount + server.SecondSubscribeCount); + } + finally { server.Release.Set(); } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task UnsubscribeWaitsForAnInFlightSubscribe(bool pattern) + { + using var server = new PausedSubscribeServer(output); + await using var connection = await server.ConnectAsync(); + 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 pending = subscription.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true, firstServer); + try + { + await server.FirstSubscribe.Task.WithTimeout(5000); + var unsubscribe = subscription.UnsubscribeFromServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, null, false); + Assert.False(unsubscribe.IsCompleted); + server.Release.Set(); + await Task.WhenAll(pending, unsubscribe).WithTimeout(5000); + await server.FirstUnsubscribe.Task.WithTimeout(5000); + Assert.Null(subscription.GetAnyCurrentServer()); + Assert.Equal(0, await subscription.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true)); + } + finally { server.Release.Set(); } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task LateAcknowledgementCannotReplaceANewerSubscription(bool pattern) + { + using var server = new PausedSubscribeServer(output); + var config = server.GetClientConfig(); + config.AsyncTimeout = 200; + config.HeartbeatInterval = TimeSpan.FromMilliseconds(20); + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + 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); + var first = subscription.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true, firstServer); + try + { + await server.FirstSubscribe.Task.WithTimeout(5000); + await Assert.ThrowsAsync(() => first.WithTimeout(5000)); + await subscription.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true, secondServer).WithTimeout(5000); + Assert.Same(secondServer, subscription.GetAnyCurrentServer()); + server.Release.Set(); + await server.FirstUnsubscribe.Task.WithTimeout(5000); + Assert.Same(secondServer, subscription.GetAnyCurrentServer()); + Assert.Equal(1, subscription.GetConnectionCount()); + } + finally { server.Release.Set(); } + } + + [Fact] + public async Task ReconnectedSubscriptionUsesTheNewPhysicalConnection() + { + using var server = new InProcessTestServer(output); + var config = server.GetClientConfig(); + config.SharedSubscriptionConnection = false; + config.HeartbeatInterval = TimeSpan.FromMilliseconds(50); + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + connection.AllowConnect = true; + var channel = RedisChannel.Literal("connection-identity"); + var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await connection.GetSubscriber().SubscribeAsync(channel, (_, value) => delivered.TrySetResult(true)); + Assert.True(connection.TryGetSubscription(channel, out var record)); + var subscription = Assert.IsType(record); + var endpoint = connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured); + var previous = endpoint.GetBridge(ConnectionType.Subscription)!.Physical!; + Assert.True(subscription.Accepts(previous)); + endpoint.SimulateConnectionFailure(SimulatedFailureType.AllSubscription); + var deadline = DateTime.UtcNow.AddSeconds(5); + while (ReferenceEquals(endpoint.GetBridge(ConnectionType.Subscription)!.Physical, previous) || !subscription.IsConnectedAny()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException("Subscription did not reconnect."); + await Task.Delay(20); + } + var current = endpoint.GetBridge(ConnectionType.Subscription)!.Physical!; + Assert.NotSame(previous, current); + Assert.False(subscription.Accepts(previous)); + Assert.True(subscription.Accepts(current)); + await connection.GetDatabase().PublishAsync(channel, "after-reconnect"); + await delivered.Task.WithTimeout(5000); + } + + 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 FirstUnsubscribe = 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)) FirstUnsubscribe.TrySetResult(true); + } + return result; + } + } +} From f43e18d79923cb96941de2cf78c631d93e5bb151 Mon Sep 17 00:00:00 2001 From: vkyow <218959617+vkyow@users.noreply.github.com> Date: Fri, 6 Nov 2026 12:23:00 +0800 Subject: [PATCH 2/3] Coalesce Sentinel discovery before starting reconfiguration --- .../ConnectionMultiplexer.Sentinel.cs | 17 +++--- .../SentinelSwitchCoalescingTests.cs | 54 +++++++++++++++++++ 2 files changed, 63 insertions(+), 8 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index dc8daba807..bfed968225 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -719,14 +719,7 @@ internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connect return false; } - // Get new primary - try twice (unless the caller already holds sentinel's answer) - EndPoint newPrimaryEndPoint = reportedPrimary - ?? GetConfiguredPrimaryForService(serviceName) - ?? GetConfiguredPrimaryForService(serviceName) - ?? throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, $"Sentinel: Failed connecting to switch primary for service: {serviceName}"); - - connection.currentSentinelPrimaryEndPoint = newPrimaryEndPoint; - + // Coalesce discovery too: simultaneous restore events otherwise fan out blocking Sentinel queries. // Claim the reconfigure gate BEFORE reshaping the connection. ReconfigureAsync refuses to run while // another reconfigure holds this gate, so rewriting the endpoint list and retiring servers first would // leave the connection reshaped but unconfigured whenever we lose that race - and two concurrent @@ -741,6 +734,14 @@ internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connect bool switched = false; try { + // Get new primary - try twice (unless the caller already holds sentinel's answer) + EndPoint newPrimaryEndPoint = reportedPrimary + ?? GetConfiguredPrimaryForService(serviceName) + ?? GetConfiguredPrimaryForService(serviceName) + ?? throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, $"Sentinel: Failed connecting to switch primary for service: {serviceName}"); + + connection.currentSentinelPrimaryEndPoint = newPrimaryEndPoint; + // Trigger a blocking reconfigure for the switch, then refresh the sentinel address list. // reconfigureAll:true forces every already-connected node to re-read its role before the // primary is re-elected (see IsStalePrimaryView); reconfigureAll:false matches the legacy diff --git a/tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs b/tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs new file mode 100644 index 0000000000..7c6288df35 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs @@ -0,0 +1,54 @@ +using System; +using System.Net; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +[RunPerProtocol] +public class SentinelSwitchCoalescingTests(ITestOutputHelper output) +{ + [Fact] + public async Task BusyReconfigurationSkipsSentinelDiscovery() + { + var primary = new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379); + var sentinelEndpoint = new IPEndPoint(IPAddress.Parse("192.0.2.10"), 26379); + using var data = new InProcessTestServer(output, primary); + using var sentinel = new InProcessSentinelServer(data, "mymaster", primary, sentinelEndpoint, output); + var config = data.GetClientConfig(); + config.EndPoints.Clear(); + config.EndPoints.Add(sentinelEndpoint); + config.ServiceName = "mymaster"; + config.Tunnel = InProcessSentinelServer.CreateTunnel(data, sentinel); + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + var owner = Assert.IsType(connection.sentinelConnection); + Assert.True(connection.TryClaimReconfigure("test in-flight switch")); + try + { + int queries = sentinel.PrimaryQueries; + for (int i = 0; i < 16; i++) Assert.False(owner.SwitchPrimary(primary, connection)); + Assert.Equal(queries, sentinel.PrimaryQueries); + } + finally { connection.ReleaseReconfigureClaim(); } + } + + [Fact] + public async Task FailedDiscoveryReleasesTheReconfigurationClaim() + { + var primary = new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379); + var sentinelEndpoint = new IPEndPoint(IPAddress.Parse("192.0.2.10"), 26379); + using var data = new InProcessTestServer(output, primary); + using var sentinel = new InProcessSentinelServer(data, "mymaster", primary, sentinelEndpoint, output); + var config = data.GetClientConfig(); + config.EndPoints.Clear(); + config.EndPoints.Add(sentinelEndpoint); + config.ServiceName = "mymaster"; + config.Tunnel = InProcessSentinelServer.CreateTunnel(data, sentinel); + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + var owner = Assert.IsType(connection.sentinelConnection); + connection.RawConfig.ServiceName = "missing"; + Assert.Throws(() => owner.SwitchPrimary(primary, connection)); + Assert.True(connection.TryClaimReconfigure("test released claim")); + connection.ReleaseReconfigureClaim(); + } +} From 65fdeb7cec75d0b5dd43e0ad0ba2d8b47576e678 Mon Sep 17 00:00:00 2001 From: vkyow <218959617+vkyow@users.noreply.github.com> Date: Wed, 11 Nov 2026 22:54:11 +0800 Subject: [PATCH 3/3] Address review feedback on subscription restoration Preserve Sentinel observations without holding the reconfigure claim during discovery. Keep the confirmed delivery route until the replacement subscribe acknowledgement, and defer unsubscribe cleanup when restoration is in flight. Complete acknowledgements even when bookkeeping fails, preserve AsyncState and subscription flags, and move cleanup dispatch outside subscription state locks. Recover sharded restores backlogged on retired nodes and add regression tests. Validated with Redis/Sentinel 8.0.3 in isolated local Docker clusters using RESP2 and RESP3: 70 controlled switches and six additional fault scenarios. No duplicate deliveries were observed with this revision; failover message loss is still possible. Focused regression tests and Windows/Linux unit suites passed. --- docs/PubSubOrder.md | 2 + .../ConnectionMultiplexer.Sentinel.cs | 32 ++- src/StackExchange.Redis/RedisSubscriber.cs | 8 +- src/StackExchange.Redis/ResultProcessor.cs | 30 ++- src/StackExchange.Redis/Subscription.cs | 214 ++++++++++++----- .../InProcessSentinelServer.cs | 3 + .../SentinelSwitchCoalescingTests.cs | 76 +++++- .../SubscriptionRestorationTests.cs | 227 +++++++++++++++++- 8 files changed, 492 insertions(+), 100 deletions(-) diff --git a/docs/PubSubOrder.md b/docs/PubSubOrder.md index d0599c16d2..cad85703e0 100644 --- a/docs/PubSubOrder.md +++ b/docs/PubSubOrder.md @@ -28,4 +28,6 @@ Reconnection Single-node subscription restoration is coordinated until Redis acknowledges the subscription. Concurrent restoration triggers cannot establish independent routes for the same logical subscription, and late acknowledgements do not replace a newer confirmed connection. This applies to callback and queue subscriptions, including pattern channels; the processing-order choices above remain unchanged. +A previously confirmed, still-live connection remains the delivery route while its replacement awaits acknowledgement. Ownership changes on the read thread when the replacement is acknowledged; later deliveries from the previous connection are rejected. Connection-state checks describe that confirmed delivery route, not a pending restore. Messages already admitted to a handler may finish after the handover. This does not make the handover atomic with publication across Redis nodes. + Redis Pub/Sub does not replay messages missed during a disconnect or failover. Work already queued for a handler may still complete after unsubscribe; applications requiring durable delivery need a replay-capable messaging design. diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs index bfed968225..329b224b26 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Sentinel.cs @@ -19,6 +19,7 @@ public partial class ConnectionMultiplexer internal ConnectionMultiplexer? sentinelConnection; private int _sentinelInitialized; private int _demotedPrimaryRecoveryActive; + private int _sentinelSwitchInProgress; /// /// Initializes the connection as a Sentinel connection and adds the necessary event handlers to track changes to the managed primaries. @@ -711,6 +712,15 @@ private static bool IsSentinelPrimaryDemoted(ConnectionMultiplexer connection) /// the view was already correct, or another reconfigure holds the gate - the callers that retry (the reconnect /// timer, the demotion heartbeat) get a later attempt either way. internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer = null, EndPoint? reportedPrimary = null) + { + // Keep a notification even if another switch or unrelated reconfigure wins the race. + if (reportedPrimary is not null) connection.currentSentinelPrimaryEndPoint = reportedPrimary; + if (Interlocked.CompareExchange(ref connection._sentinelSwitchInProgress, 1, 0) != 0) return false; + try { return SwitchPrimaryCore(switchBlame, connection, writer, reportedPrimary); } + finally { Volatile.Write(ref connection._sentinelSwitchInProgress, 0); } + } + + private bool SwitchPrimaryCore(EndPoint? switchBlame, ConnectionMultiplexer connection, TextWriter? writer, EndPoint? reportedPrimary) { var logger = Logger.With(writer); if (connection.RawConfig.ServiceName is not string serviceName) @@ -719,11 +729,15 @@ internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connect return false; } - // Coalesce discovery too: simultaneous restore events otherwise fan out blocking Sentinel queries. - // Claim the reconfigure gate BEFORE reshaping the connection. ReconfigureAsync refuses to run while - // another reconfigure holds this gate, so rewriting the endpoint list and retiring servers first would - // leave the connection reshaped but unconfigured whenever we lose that race - and two concurrent - // switches could interleave their endpoint rewrites. Claim first and a lost race touches nothing. + // Discovery and recording Sentinel's answer do not reshape the connection. Keep blocking + // queries outside the claim, and retain the reported primary even if another reconfigure wins. + EndPoint newPrimaryEndPoint = reportedPrimary + ?? GetConfiguredPrimaryForService(serviceName) + ?? GetConfiguredPrimaryForService(serviceName) + ?? throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, $"Sentinel: Failed connecting to switch primary for service: {serviceName}"); + connection.currentSentinelPrimaryEndPoint = newPrimaryEndPoint; + + // Claim before rewriting endpoints: a lost claim must not leave a half-reconfigured topology. string cause = $"Primary switch {serviceName}"; if (!connection.TryClaimReconfigure(cause)) { @@ -734,14 +748,6 @@ internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connect bool switched = false; try { - // Get new primary - try twice (unless the caller already holds sentinel's answer) - EndPoint newPrimaryEndPoint = reportedPrimary - ?? GetConfiguredPrimaryForService(serviceName) - ?? GetConfiguredPrimaryForService(serviceName) - ?? throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, $"Sentinel: Failed connecting to switch primary for service: {serviceName}"); - - connection.currentSentinelPrimaryEndPoint = newPrimaryEndPoint; - // Trigger a blocking reconfigure for the switch, then refresh the sentinel address list. // reconfigureAll:true forces every already-connected node to re-read its role before the // primary is re-elected (see IsStalePrimaryView); reconfigureAll:false matches the legacy diff --git a/src/StackExchange.Redis/RedisSubscriber.cs b/src/StackExchange.Redis/RedisSubscriber.cs index 0becb4dc60..1ee33d5d55 100644 --- a/src/StackExchange.Redis/RedisSubscriber.cs +++ b/src/StackExchange.Redis/RedisSubscriber.cs @@ -84,7 +84,7 @@ internal void OnMessage(in RedisChannel subscription, in RedisChannel channel, i ChannelMessageQueue? queues = null; if (subscriptions.TryGetValue(subscription, out Subscription? sub)) { - if (source is not null && sub is SingleNodeSubscription single && !single.Accepts(source)) return; + if (source is not null && !sub.Accepts(source)) return; completable = sub.ForInvoke(channel, payload, out queues); } if (queues != null) @@ -266,10 +266,10 @@ internal void ResubscribeToServer(Subscription sub, in RedisChannel channel, Ser { // we'll *try* for a simple resubscribe, following any -MOVED etc, but if that fails: fall back // to full reconfigure; importantly, note that we've already recorded the disconnect - var message = sub.GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, CommandFlags.None, false); - _ = (sub is SingleNodeSubscription + var restore = sub is SingleNodeSubscription ? (Task)sub.EnsureSubscribedToServerAsync(this, channel, CommandFlags.None, false, serverEndPoint) - : ExecuteAsync(message, sub.Processor, serverEndPoint)).ContinueWith( + : ExecuteAsync(sub.GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, CommandFlags.None, false), sub.Processor, serverEndPoint); + _ = restore.ContinueWith( t => multiplexer.ReconfigureIfNeeded(serverEndPoint.EndPoint, false, cause: cause), TaskContinuationOptions.OnlyOnFaulted); } diff --git a/src/StackExchange.Redis/ResultProcessor.cs b/src/StackExchange.Redis/ResultProcessor.cs index ec2fe82d48..a3b6d9968b 100644 --- a/src/StackExchange.Redis/ResultProcessor.cs +++ b/src/StackExchange.Redis/ResultProcessor.cs @@ -558,23 +558,27 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes if (iter.Value.TryReadInt64(out long count)) { connection.SubscriptionCount = count; - var ep = connection.BridgeCouldBeNull?.ServerEndPoint; - if (ep is not null) + try { - switch (message.Command) + var ep = connection.BridgeCouldBeNull?.ServerEndPoint; + if (ep is not null) { - case RedisCommand.SUBSCRIBE: - case RedisCommand.SSUBSCRIBE: - case RedisCommand.PSUBSCRIBE: - if (_onSubscribed is not null) _onSubscribed(connection); - else Subscription?.AddEndpoint(ep); - break; - default: - Subscription?.TryRemoveEndpoint(ep); - break; + switch (message.Command) + { + case RedisCommand.SUBSCRIBE: + case RedisCommand.SSUBSCRIBE: + case RedisCommand.PSUBSCRIBE: + if (_onSubscribed is not null) _onSubscribed(connection); + else Subscription?.AddEndpoint(ep); + break; + default: + Subscription?.TryRemoveEndpoint(ep); + break; + } } } - SetResult(message, true); + catch (Exception ex) { connection.BridgeCouldBeNull?.Multiplexer.OnInternalError(ex); } + finally { SetResult(message, true); } return true; } } diff --git a/src/StackExchange.Redis/Subscription.cs b/src/StackExchange.Redis/Subscription.cs index 1e70c3b7a8..15025fcecd 100644 --- a/src/StackExchange.Redis/Subscription.cs +++ b/src/StackExchange.Redis/Subscription.cs @@ -22,6 +22,8 @@ internal abstract class Subscription public CommandFlags Flags { get; } public ResultProcessor.TrackSubscriptionsProcessor Processor { get; } + internal virtual bool Accepts(PhysicalConnection connection) => true; + internal abstract bool IsConnectedAny(); internal abstract bool IsConnectedTo(EndPoint endpoint); @@ -206,18 +208,20 @@ internal sealed class SingleNodeSubscription(CommandFlags flags) : Subscription( private RestoreOperation? _pendingRestore; private System.Collections.Generic.List? _cleanup; private bool _unsubscribed; + private bool _needsRestore; internal override bool IsConnectedAny() { var connection = Volatile.Read(ref _currentConnection); - return _currentServer is { IsSubscriberConnected: true } - && (connection is null || ReferenceEquals(connection.BridgeCouldBeNull?.Physical, connection)); + return !Volatile.Read(ref _unsubscribed) + && _currentServer is { IsSubscriberConnected: true } + && connection is not null && ReferenceEquals(connection.BridgeCouldBeNull?.Physical, connection); } internal override int GetConnectionCount() => IsConnectedAny() ? 1 : 0; internal override bool IsConnectedTo(EndPoint endpoint) - => _currentServer is { IsSubscriberConnected: true } server && server.EndPoint == endpoint; + => IsConnectedAny() && _currentServer?.EndPoint == endpoint; - internal bool Accepts(PhysicalConnection connection) + internal override bool Accepts(PhysicalConnection connection) => !Volatile.Read(ref _unsubscribed) && ReferenceEquals(Volatile.Read(ref _currentConnection), connection); internal override void AddEndpoint(ServerEndPoint server) { lock (_stateLock) _currentServer = server; } @@ -227,8 +231,12 @@ internal override bool TryRemoveEndpoint(ServerEndPoint expected) lock (_stateLock) { if (_currentServer != expected) return false; - _currentServer = null; - Volatile.Write(ref _currentConnection, null); + _needsRestore = true; + if (!IsConnectedAny()) + { + _currentServer = null; + Volatile.Write(ref _currentConnection, null); + } return true; } } @@ -238,6 +246,7 @@ internal override bool TryRemoveEndpoint(ServerEndPoint expected) internal override void RemoveDisconnectedEndpoints() { + RestoreOperation? superseded = null; lock (_stateLock) { if (_currentServer is not null && !IsConnectedAny()) @@ -245,12 +254,24 @@ internal override void RemoveDisconnectedEndpoints() _currentServer = null; Volatile.Write(ref _currentConnection, null); } + // A redirected sharded restore can be backlogged on a retired node. Once the + // topology has a live owner, don't hold every later restore behind that old reply. + if (_pendingRestore is { } pending && pending.Channel.IsKeyRouted && pending.Message.IsBacklogged) + { + var muxer = pending.Subscriber.multiplexer; + var selected = muxer.SelectServer(pending.Message); + if (selected is { IsSubscriberConnected: true } + && muxer.ServerSelectionStrategy.CanServeSlot(selected, pending.Channel)) + superseded = pending; + } } + superseded?.Supersede(); } internal override int EnsureSubscribedToServer(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall) { - bool fireAndForget = (flags & CommandFlags.FireAndForget) != 0; + RemoveDisconnectedEndpoints(); + bool fireAndForget = ((Flags | flags) & CommandFlags.FireAndForget) != 0; if (!Enter(subscriber, fireAndForget)) return 0; RestoreOperation? operation = null; bool transferred = false; @@ -275,7 +296,8 @@ internal override int EnsureSubscribedToServer(RedisSubscriber subscriber, in Re internal override async Task EnsureSubscribedToServerAsync(RedisSubscriber subscriber, RedisChannel channel, CommandFlags flags, bool internalCall, ServerEndPoint? server = null) { - bool fireAndForget = (flags & CommandFlags.FireAndForget) != 0; + RemoveDisconnectedEndpoints(); + bool fireAndForget = ((Flags | flags) & CommandFlags.FireAndForget) != 0; if (fireAndForget) { if (!Enter(subscriber, true)) return 0; @@ -296,7 +318,7 @@ internal override async Task EnsureSubscribedToServerAsync(RedisSubscriber _ = CompleteRestoreAsync(operation).ObserveErrors(); return 1; } - await subscriber.ExecuteAsync(operation.Message, operation.Processor, operation.Server).ForAwait(); + await AwaitRestoreAsync(operation).ForAwait(); return 1; } finally { if (!transferred) EndOperation(operation); } @@ -317,10 +339,12 @@ private Exception WaitingTimeout(RedisSubscriber subscriber) lock (_stateLock) { if (_unsubscribed) return null; - RemoveIncorrectRouting(subscriber, channel, flags, internalCall); - if (IsConnectedAny()) return null; - _currentServer = null; - Volatile.Write(ref _currentConnection, null); + bool incorrectRouting = channel.IsKeyRouted + && _currentServer is { ServerType: ServerType.Cluster } current + && !subscriber.multiplexer.ServerSelectionStrategy.CanServeSlot(current, channel); + if (!_needsRestore && !incorrectRouting && IsConnectedAny()) return null; + // Keep the previous delivery route until the replacement is acknowledged. + // The read thread switches ownership before processing the next incoming message. // Keep an actual reply task even for fire-and-forget restoration: sending is not acknowledgement. var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall, trackReply: true); server ??= subscriber.multiplexer.SelectServer(message); @@ -330,38 +354,60 @@ private Exception WaitingTimeout(RedisSubscriber subscriber) private async Task CompleteRestoreAsync(RestoreOperation operation) { - try { await operation.Subscriber.ExecuteAsync(operation.Message, operation.Processor, operation.Server).ForAwait(); } + try { await AwaitRestoreAsync(operation).ForAwait(); } finally { EndOperation(operation); } } + private static async Task AwaitRestoreAsync(RestoreOperation operation) + { + var reply = operation.Subscriber.ExecuteAsync(operation.Message, operation.Processor, operation.Server); + if (await Task.WhenAny(reply, operation.Superseded).ForAwait() != reply) + { + _ = reply.ObserveErrors(); + throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, "The subscription restore was superseded by a live routing owner."); + } + await reply.ForAwait(); + } + private void EndOperation(RestoreOperation? operation) { - try + lock (_stateLock) { - lock (_stateLock) - { - if (ReferenceEquals(_pendingRestore, operation)) _pendingRestore = null; - DrainCleanup(); - } + if (ReferenceEquals(_pendingRestore, operation)) _pendingRestore = null; + _operationLock.Release(); + } + DrainCleanup(); + if (operation is not null && operation.Superseded.IsCompleted && !_unsubscribed && !IsConnectedAny()) + { + var muxer = operation.Subscriber.multiplexer; + if (muxer.TryGetSubscription(operation.Channel, out var active) && ReferenceEquals(active, this) + && muxer.SelectServer(operation.Message) is { IsSubscriberConnected: true } selected) + operation.Subscriber.ResubscribeToServer(this, operation.Channel, selected, "superseded sharded restoration"); } - finally { _operationLock.Release(); } } private void OnSubscribed(RestoreOperation operation, PhysicalConnection connection) { var server = connection.BridgeCouldBeNull?.ServerEndPoint; if (server is null) return; + CleanupRequest? cleanup = null; lock (_stateLock) { - if (ReferenceEquals(_pendingRestore, operation)) + if (ReferenceEquals(_pendingRestore, operation) && !_unsubscribed && !operation.Superseded.IsCompleted) { + if (_currentServer is { } previous && !ReferenceEquals(previous, server)) + cleanup = new CleanupRequest(operation.Subscriber, operation.Channel, previous); _currentServer = server; Volatile.Write(ref _currentConnection, connection); - return; + _needsRestore = false; + } + else + { + // An obsolete acknowledgement must not change the confirmed delivery route. + cleanup = new CleanupRequest(operation.Subscriber, operation.Channel, server); } - // A timed-out request can still be acknowledged. It must not replace a newer route. - QueueCleanup(new CleanupRequest(operation.Subscriber, operation.Channel, server)); } + if (cleanup is not null) QueueCleanup(cleanup); } private void QueueCleanup(CleanupRequest request) @@ -369,47 +415,82 @@ private void QueueCleanup(CleanupRequest request) lock (_stateLock) { (_cleanup ??= new()).Add(request); - DrainCleanup(); } + DrainCleanup(); } private void DrainCleanup() { - // A redirected restore may not have its final endpoint yet. Reconcile after its acknowledgement. - if (_pendingRestore is not null || _cleanup is null) return; - var cleanup = _cleanup; - _cleanup = null; - foreach (var request in cleanup) + // Serialize cleanup writes with restoration, but never write or forward under _stateLock. + if (!_operationLock.Wait(0)) return; + bool ownsOperation = true; + try { - var muxer = request.Subscriber.multiplexer; - if (muxer.TryGetSubscription(request.Channel, out var active) && !ReferenceEquals(active, this)) + while (true) { - if (active is SingleNodeSubscription single) + System.Collections.Generic.List? cleanup; + lock (_stateLock) { - single.QueueCleanup(request); - continue; + cleanup = _cleanup; + _cleanup = null; + if (cleanup is null) + { + // QueueCleanup cannot miss this release: it takes the same state lock. + _operationLock.Release(); + ownsOperation = false; + return; + } + } + foreach (var request in cleanup) + { + var muxer = request.Subscriber.multiplexer; + if (muxer.TryGetSubscription(request.Channel, out var active) && !ReferenceEquals(active, this)) + { + if (active is SingleNodeSubscription single) + { + single.QueueCleanup(request); + continue; + } + if (active.IsConnectedTo(request.Server.EndPoint)) continue; + } + if (!_unsubscribed && ReferenceEquals(_currentServer, request.Server)) continue; + try + { + var message = GetSubscriptionMessage(request.Channel, SubscriptionAction.Unsubscribe, CommandFlags.FireAndForget, true); + muxer.ExecuteAsyncImpl(message, ResultProcessor.TrackSubscriptions, null, request.Server).ObserveErrors(); + } + catch (Exception ex) { muxer.OnInternalError(ex); } } - if (active.IsConnectedTo(request.Server.EndPoint)) continue; - } - if (!_unsubscribed && ReferenceEquals(_currentServer, request.Server)) continue; - try - { - var message = GetSubscriptionMessage(request.Channel, SubscriptionAction.Unsubscribe, CommandFlags.FireAndForget, true); - muxer.ExecuteAsyncImpl(message, ResultProcessor.TrackSubscriptions, null, request.Server).ObserveErrors(); } - catch (Exception ex) { muxer.OnInternalError(ex); } } + finally { if (ownsOperation) _operationLock.Release(); } + } + + private void QueueUnsubscribeCleanup(RedisSubscriber subscriber, RedisChannel channel) + { + lock (_stateLock) + { + if (_currentServer is { } current) + (_cleanup ??= new()).Add(new CleanupRequest(subscriber, channel, current)); + if (_pendingRestore?.Server is { } pending && !ReferenceEquals(pending, _currentServer)) + (_cleanup ??= new()).Add(new CleanupRequest(subscriber, channel, pending)); + } + DrainCleanup(); } internal override bool UnsubscribeFromServer(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall) { Volatile.Write(ref _unsubscribed, true); - if ((flags & CommandFlags.FireAndForget) != 0) + if (((Flags | flags) & CommandFlags.FireAndForget) != 0) { UnsubscribeCoreAsync(subscriber, channel, flags, null, internalCall).ObserveErrors(); return true; } - Enter(subscriber, false); + if (!_operationLock.Wait(subscriber.multiplexer.TimeoutMilliseconds)) + { + QueueUnsubscribeCleanup(subscriber, channel); + return true; + } try { var server = _currentServer; @@ -425,14 +506,30 @@ internal override Task UnsubscribeFromServerAsync(RedisSubscriber subscrib { Volatile.Write(ref _unsubscribed, true); var task = UnsubscribeCoreAsync(subscriber, channel, flags, asyncState, internalCall); - if ((flags & CommandFlags.FireAndForget) == 0) return task; + if (((Flags | flags) & CommandFlags.FireAndForget) == 0) + { + var completion = TaskSource.Create(asyncState, TaskCreationOptions.RunContinuationsAsynchronously); + CompleteUnsubscribeAsync(task, completion).ObserveErrors(); + return completion.Task; + } task.ObserveErrors(); return CompletedTask.FromResult(true, asyncState); } + private static async Task CompleteUnsubscribeAsync(Task operation, TaskCompletionSource completion) + { + try { completion.TrySetResult(await operation.ForAwait()); } + catch (OperationCanceledException) when (operation.IsCanceled) { completion.TrySetCanceled(); } + catch (Exception ex) { completion.TrySetException(ex); } + } + private async Task UnsubscribeCoreAsync(RedisSubscriber subscriber, RedisChannel channel, CommandFlags flags, object? asyncState, bool internalCall) { - if (!await _operationLock.WaitAsync(subscriber.multiplexer.AsyncTimeoutMilliseconds).ForAwait()) throw WaitingTimeout(subscriber); + if (!await _operationLock.WaitAsync(subscriber.multiplexer.AsyncTimeoutMilliseconds).ForAwait()) + { + QueueUnsubscribeCleanup(subscriber, channel); + return true; + } try { var server = _currentServer; @@ -457,6 +554,10 @@ private bool ForwardCleanup(RedisSubscriber subscriber, RedisChannel channel, Se private sealed class RestoreOperation { + private readonly TaskCompletionSource _superseded = TaskSource.Create(null, TaskCreationOptions.RunContinuationsAsynchronously); + public Task Superseded => _superseded.Task; + public void Supersede() => _superseded.TrySetResult(true); + public RestoreOperation(SingleNodeSubscription owner, RedisSubscriber subscriber, RedisChannel channel, Message message, ServerEndPoint? server) { Subscriber = subscriber; @@ -478,23 +579,6 @@ private sealed class CleanupRequest(RedisSubscriber subscriber, RedisChannel cha public RedisChannel Channel { get; } = channel; public ServerEndPoint Server { get; } = server; } - - 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 - // a single-key keyspace notification); is the subscribed server still handling that channel? - if (channel.IsKeyRouted && _currentServer is { ServerType: ServerType.Cluster } current) - { - // if we consider replicas, there can be multiple valid target servers; we can't ask - // "is this the correct server?", but we can ask "is it suitable?", based on the slot - if (!subscriber.multiplexer.ServerSelectionStrategy.CanServeSlot(_currentServer, channel)) - { - var message = GetSubscriptionMessage(channel, SubscriptionAction.Unsubscribe, flags | CommandFlags.FireAndForget, internalCall); - subscriber.multiplexer.ExecuteSyncImpl(message, Processor, current); - _currentServer = null; // pre-emptively disconnect - F+F - } - } - } } // used for keyspace subscriptions, which are routed to multiple nodes diff --git a/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs b/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs index 8fecb98e69..8e98b7d418 100644 --- a/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs +++ b/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs @@ -51,12 +51,15 @@ public EndPoint Primary private int _primaryQueries; + internal Action? BeforePrimaryQuery { get; set; } + /// How many times a client has asked this sentinel for the primary. public int PrimaryQueries => Volatile.Read(ref _primaryQueries); [RedisCommand(3, nameof(RedisCommand.SENTINEL), "get-master-addr-by-name", LockFree = true)] private TypedRedisValue SentinelGetMasterAddrByName(RedisClient client, in RedisRequest request) { + BeforePrimaryQuery?.Invoke(); Interlocked.Increment(ref _primaryQueries); if (!IsService(request)) return TypedRedisValue.NullArray(RespPrefix.Array); var host = GetHost(Primary, out var port); diff --git a/tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs b/tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs index 7c6288df35..2f5a444fbd 100644 --- a/tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs +++ b/tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs @@ -1,5 +1,6 @@ using System; using System.Net; +using System.Threading; using System.Threading.Tasks; using Xunit; @@ -9,7 +10,7 @@ namespace StackExchange.Redis.Tests; public class SentinelSwitchCoalescingTests(ITestOutputHelper output) { [Fact] - public async Task BusyReconfigurationSkipsSentinelDiscovery() + public async Task BusyReconfigurationStillRecordsTheReportedPrimary() { var primary = new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379); var sentinelEndpoint = new IPEndPoint(IPAddress.Parse("192.0.2.10"), 26379); @@ -26,12 +27,83 @@ public async Task BusyReconfigurationSkipsSentinelDiscovery() try { int queries = sentinel.PrimaryQueries; - for (int i = 0; i < 16; i++) Assert.False(owner.SwitchPrimary(primary, connection)); + var reported = new IPEndPoint(IPAddress.Parse("192.0.2.2"), 6379); + for (int i = 0; i < 16; i++) Assert.False(owner.SwitchPrimary(primary, connection, reportedPrimary: reported)); Assert.Equal(queries, sentinel.PrimaryQueries); + Assert.Equal(reported, connection.currentSentinelPrimaryEndPoint); } finally { connection.ReleaseReconfigureClaim(); } } + [Fact] + public async Task DiscoveryDoesNotHoldTheReconfigurationClaim() + { + var primary = new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379); + var sentinelEndpoint = new IPEndPoint(IPAddress.Parse("192.0.2.10"), 26379); + using var data = new InProcessTestServer(output, primary); + using var sentinel = new InProcessSentinelServer(data, "mymaster", primary, sentinelEndpoint, output); + var config = data.GetClientConfig(); + config.EndPoints.Clear(); + config.EndPoints.Add(sentinelEndpoint); + config.ServiceName = "mymaster"; + config.Tunnel = InProcessSentinelServer.CreateTunnel(data, sentinel); + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + var owner = Assert.IsType(connection.sentinelConnection); + bool sawQuery = false, claimAvailable = false; + connection.RawConfig.ServiceName = "missing"; + sentinel.BeforePrimaryQuery = () => + { + sawQuery = true; + claimAvailable = connection.TryClaimReconfigure("test during discovery"); + if (claimAvailable) connection.ReleaseReconfigureClaim(); + }; + try { Assert.Throws(() => owner.SwitchPrimary(primary, connection)); } + finally { sentinel.BeforePrimaryQuery = null; } + Assert.True(sawQuery); + Assert.True(claimAvailable); + } + + [Fact] + public async Task SlowDiscoveryCoalescesSwitchesWithoutHoldingTheReconfigurationClaim() + { + var primary = new IPEndPoint(IPAddress.Parse("192.0.2.1"), 6379); + var sentinelEndpoint = new IPEndPoint(IPAddress.Parse("192.0.2.10"), 26379); + using var data = new InProcessTestServer(output, primary); + using var sentinel = new InProcessSentinelServer(data, "mymaster", primary, sentinelEndpoint, output); + var config = data.GetClientConfig(); + config.EndPoints.Clear(); + config.EndPoints.Add(sentinelEndpoint); + config.ServiceName = "mymaster"; + config.Tunnel = InProcessSentinelServer.CreateTunnel(data, sentinel); + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + var owner = Assert.IsType(connection.sentinelConnection); + connection.RawConfig.ServiceName = "missing"; + using var release = new ManualResetEventSlim(); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + sentinel.BeforePrimaryQuery = () => + { + entered.TrySetResult(true); + if (!release.Wait(TimeSpan.FromSeconds(10))) throw new TimeoutException("Sentinel discovery barrier timed out."); + }; + int before = sentinel.PrimaryQueries; + var pending = Task.Run(() => Assert.Throws(() => owner.SwitchPrimary(primary, connection))); + try + { + await entered.Task.WithTimeout(5000); + Assert.True(connection.TryClaimReconfigure("test unrelated reconfiguration")); + connection.ReleaseReconfigureClaim(); + for (int i = 0; i < 16; i++) Assert.False(owner.SwitchPrimary(primary, connection, reportedPrimary: primary)); + Assert.Equal(before, sentinel.PrimaryQueries); + Assert.Equal(primary, connection.currentSentinelPrimaryEndPoint); + } + finally + { + release.Set(); + await pending.WithTimeout(5000); + sentinel.BeforePrimaryQuery = null; + } + } + [Fact] public async Task FailedDiscoveryReleasesTheReconfigurationClaim() { diff --git a/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs b/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs index cb7a0a9a52..9413a6e37c 100644 --- a/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs +++ b/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs @@ -36,6 +36,7 @@ public async Task OverlappingRestoresCannotSubscribeOnTwoServers(bool pattern, b Assert.Equal(0, Volatile.Read(ref server.SecondSubscribeCount)); server.Release.Set(); await Task.WhenAll(first, second).WithTimeout(5000); + Assert.True(await Poll.UntilAsync(() => subscription.IsConnectedAny(), timeoutMilliseconds: 5000)); Assert.Equal(1, Volatile.Read(ref server.FirstSubscribeCount)); Assert.Equal(0, Volatile.Read(ref server.SecondSubscribeCount)); Assert.Same(firstServer, subscription.GetAnyCurrentServer()); @@ -155,12 +156,226 @@ public async Task ReconnectedSubscriptionUsesTheNewPhysicalConnection() await delivered.Task.WithTimeout(5000); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task TimedOutSynchronousUnsubscribeStillDetachesThePendingSubscription(bool pattern) + { + using var server = new PausedSubscribeServer(output); + var config = server.GetClientConfig(); + config.SyncTimeout = 100; + config.AsyncTimeout = 5000; + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + var subscriber = connection.GetSubscriber(); + var channel = pattern ? RedisChannel.Pattern("restoration.*") : RedisChannel.Literal("restoration.channel"); + var subscription = connection.GetOrAddSubscription(channel, CommandFlags.None); + subscription.Add((_, _) => { }, null); + var pending = subscription.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, false, + connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured)); + try + { + await server.FirstSubscribe.Task.WithTimeout(5000); + subscriber.Unsubscribe(channel); + Assert.False(connection.TryGetSubscription(channel, out _)); + server.Release.Set(); + await pending.WithTimeout(5000); + await server.FirstUnsubscribe.Task.WithTimeout(5000); + } + finally { server.Release.Set(); } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task UnsubscribeAllContinuesPastAnInFlightSubscription(bool pattern) + { + using var server = new PausedSubscribeServer(output); + var config = server.GetClientConfig(); + config.SyncTimeout = 100; + config.AsyncTimeout = 5000; + await using var connection = await ConnectionMultiplexer.ConnectAsync(config); + var subscriber = connection.DefaultSubscriber; + var firstChannel = pattern ? RedisChannel.Pattern("restoration.first.*") : RedisChannel.Literal("restoration.first"); + var secondChannel = pattern ? RedisChannel.Pattern("restoration.second.*") : RedisChannel.Literal("restoration.second"); + var first = connection.GetOrAddSubscription(firstChannel, CommandFlags.None); + var second = new ConnectionMultiplexer.SingleNodeSubscription(CommandFlags.None); + var pending = first.EnsureSubscribedToServerAsync(subscriber, firstChannel, CommandFlags.None, false, + connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured)); + try + { + await server.FirstSubscribe.Task.WithTimeout(5000); + await second.EnsureSubscribedToServerAsync(subscriber, secondChannel, CommandFlags.None, false, + connection.GetServerEndPoint(server.SecondEndPoint, ServerProvenance.Configured)).WithTimeout(5000); + Assert.True(connection.GetSubscriptions().TryAdd(secondChannel, second)); + subscriber.UnsubscribeAll(); + Assert.False(connection.TryGetSubscription(firstChannel, out _)); + Assert.False(connection.TryGetSubscription(secondChannel, out _)); + server.Release.Set(); + await pending.WithTimeout(5000); + Assert.True(await Poll.UntilAsync(() => Volatile.Read(ref server.UnsubscribeCount) >= 2, timeoutMilliseconds: 5000)); + } + finally { server.Release.Set(); } + } + + [Theory] + [InlineData(false, false)] + [InlineData(true, false)] + [InlineData(false, true)] + [InlineData(true, true)] + public async Task SubscriptionFireAndForgetFlagsApplyToLaterRestoreCalls(bool pattern, bool asynchronous) + { + 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.FireAndForget); + var endpoint = connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured); + await subscription.EnsureSubscribedToServerAsync(subscriber, channel, CommandFlags.FireAndForget, true, endpoint); + try + { + await server.FirstSubscribe.Task.WithTimeout(5000); + if (asynchronous) + Assert.Equal(0, await subscription.EnsureSubscribedToServerAsync(subscriber, channel, CommandFlags.None, true).WithTimeout(1000)); + else + Assert.Equal(0, subscription.EnsureSubscribedToServer(subscriber, channel, CommandFlags.None, true)); + } + finally { server.Release.Set(); } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task LiveRouteDeliversUntilTheReplacementAcknowledgement(bool pattern) + { + using var server = new PausedSubscribeServer(output); + server.Release.Set(); + await using var connection = await server.ConnectAsync(); + var channel = pattern ? RedisChannel.Pattern("restoration.*") : RedisChannel.Literal("restoration.channel"); + var subscription = Assert.IsType(connection.GetOrAddSubscription(channel, CommandFlags.None)); + var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + subscription.Add((_, value) => { if (value == "during-restore") delivered.TrySetResult(true); }, null); + var first = connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured); + var second = connection.GetServerEndPoint(server.SecondEndPoint, ServerProvenance.Configured); + await subscription.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true, first); + var physical = first.GetBridge(ConnectionType.Subscription)!.Physical!; + server.PauseSecond = true; + server.Release.Reset(); + Assert.True(subscription.TryRemoveEndpoint(first)); + var pending = subscription.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true, second); + try + { + await server.SecondSubscribe.Task.WithTimeout(5000); + Assert.True(subscription.IsConnectedAny()); + Assert.True(subscription.Accepts(physical)); + Assert.Same(first, subscription.GetAnyCurrentServer()); + await connection.GetDatabase().PublishAsync(RedisChannel.Literal("restoration.channel"), "during-restore"); + await delivered.Task.WithTimeout(5000); + server.Release.Set(); + await pending.WithTimeout(5000); + Assert.False(subscription.Accepts(physical)); + Assert.True(subscription.Accepts(second.GetBridge(ConnectionType.Subscription)!.Physical!)); + await server.FirstUnsubscribe.Task.WithTimeout(5000); + } + finally { server.Release.Set(); } + } + + [Fact] + public async Task SubscribeAcknowledgementCompletesWhenBookkeepingThrows() + { + using var server = new InProcessTestServer(output); + await using var connection = await server.ConnectAsync(); + var reported = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + connection.InternalError += (_, e) => { if (e.Exception is InvalidOperationException) reported.TrySetResult(true); }; + var subscription = new ConnectionMultiplexer.SingleNodeSubscription(CommandFlags.None); + var message = subscription.GetSubscriptionMessage(RedisChannel.Literal("ack-failure"), ConnectionMultiplexer.SubscriptionAction.Subscribe, + CommandFlags.None, false, trackReply: true); + var processor = new ResultProcessor.TrackSubscriptionsProcessor(subscription, _ => throw new InvalidOperationException("test bookkeeping failure")); + var endpoint = connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured); + Assert.True(await connection.ExecuteAsyncImpl(message, processor, null, endpoint).WithTimeout(5000)); + await reported.Task.WithTimeout(5000); + await connection.GetSubscriber().PingAsync().WithTimeout(5000); + } + + [Fact] + public async Task UnsubscribeTaskPreservesTheCallersAsyncState() + { + using var server = new InProcessTestServer(output); + await using var connection = await server.ConnectAsync(); + var state = new object(); + var subscriber = connection.GetSubscriber(state); + var channel = RedisChannel.Literal("async-state"); + await subscriber.SubscribeAsync(channel, (_, _) => { }); + var pending = subscriber.UnsubscribeAsync(channel); + Assert.Same(state, pending.AsyncState); + await pending.WithTimeout(5000); + } + + [Fact] + public async Task LateCleanupDoesNotUnsubscribeTheReplacementOwner() + { + using var server = new PausedSubscribeServer(output); + await using var connection = await server.ConnectAsync(); + var channel = RedisChannel.Literal("restoration.owner"); + var old = connection.GetOrAddSubscription(channel, CommandFlags.None); + var first = connection.GetServerEndPoint(server.DefaultEndPoint, ServerProvenance.Configured); + var second = connection.GetServerEndPoint(server.SecondEndPoint, ServerProvenance.Configured); + var pending = old.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true, first); + try + { + await server.FirstSubscribe.Task.WithTimeout(5000); + await connection.GetSubscriber().UnsubscribeAsync(channel, flags: CommandFlags.FireAndForget); + var replacement = new ConnectionMultiplexer.SingleNodeSubscription(CommandFlags.None); + var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + replacement.Add((_, value) => { if (value == "new-owner") received.TrySetResult(true); }, null); + await replacement.EnsureSubscribedToServerAsync(connection.DefaultSubscriber, channel, CommandFlags.None, true, second); + Assert.True(connection.GetSubscriptions().TryAdd(channel, replacement)); + server.Release.Set(); + await pending.WithTimeout(5000); + await server.FirstUnsubscribe.Task.WithTimeout(5000); + Assert.True(replacement.IsConnectedAny()); + Assert.Same(second, replacement.GetAnyCurrentServer()); + await connection.GetDatabase().PublishAsync(channel, "new-owner"); + await received.Task.WithTimeout(5000); + } + finally { server.Release.Set(); } + } + + [Fact] + public async Task RepeatedSubscriptionReplacementDoesNotCallStoppedHandlers() + { + using var server = new InProcessTestServer(output); + await using var connection = await server.ConnectAsync(); + var subscriber = connection.GetSubscriber(); + var channel = RedisChannel.Literal("repeated-owner"); + int deliveries = 0, stoppedDeliveries = 0; + for (int i = 0; i < 25; i++) + { + int stopped = 0; + var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Action handler = (_, _) => + { + if (Volatile.Read(ref stopped) != 0) Interlocked.Increment(ref stoppedDeliveries); + Interlocked.Increment(ref deliveries); + received.TrySetResult(true); + }; + await subscriber.SubscribeAsync(channel, handler); + await subscriber.PublishAsync(channel, i); + await received.Task.WithTimeout(5000); + await subscriber.UnsubscribeAsync(channel, handler); + Volatile.Write(ref stopped, 1); + } + Assert.Equal(25, Volatile.Read(ref deliveries)); + Assert.Equal(0, Volatile.Read(ref stoppedDeliveries)); + } + 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 int FirstSubscribeCount, SecondSubscribeCount, UnsubscribeCount; + public bool PauseSecond; + public readonly TaskCompletionSource SecondSubscribe = new(TaskCreationOptions.RunContinuationsAsynchronously); public readonly TaskCompletionSource FirstUnsubscribe = new(TaskCreationOptions.RunContinuationsAsynchronously); public PausedSubscribeServer(ITestOutputHelper output) : base(output) => SecondEndPoint = AddEmptyNode(); @@ -175,15 +390,21 @@ public override TypedRedisValue Execute(RedisClient client, in RedisRequest requ { Interlocked.Increment(ref FirstSubscribeCount); FirstSubscribe.TrySetResult(true); - if (!Release.Wait(TimeSpan.FromSeconds(10))) throw new TimeoutException("Test subscription barrier timed out."); + if (!PauseSecond && !Release.Wait(TimeSpan.FromSeconds(10))) throw new TimeoutException("Test subscription barrier timed out."); + } + else + { + Interlocked.Increment(ref SecondSubscribeCount); + SecondSubscribe.TrySetResult(true); + if (PauseSecond && !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); + Interlocked.Increment(ref UnsubscribeCount); if (ReferenceEquals(client.Node, firstNode)) FirstUnsubscribe.TrySetResult(true); } return result;