diff --git a/docs/PubSubOrder.md b/docs/PubSubOrder.md
index 9a7104962..cad85703e 100644
--- a/docs/PubSubOrder.md
+++ b/docs/PubSubOrder.md
@@ -23,3 +23,11 @@ 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.
+
+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 dc8daba80..329b224b2 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,18 +729,15 @@ internal bool SwitchPrimary(EndPoint? switchBlame, ConnectionMultiplexer connect
return false;
}
- // Get new primary - try twice (unless the caller already holds sentinel's answer)
+ // 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 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.
+ // Claim before rewriting endpoints: a lost claim must not leave a half-reconfigured topology.
string cause = $"Primary switch {serviceName}";
if (!connection.TryClaimReconfigure(cause))
{
diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs
index 5b572c471..190f3f183 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 d0741e5f5..1ee33d5d5 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.Accepts(source)) return;
completable = sub.ForInvoke(channel, payload, out queues);
}
if (queues != null)
@@ -262,8 +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);
- _ = ExecuteAsync(message, sub.Processor, serverEndPoint).ContinueWith(
+ var restore = sub is SingleNodeSubscription
+ ? (Task)sub.EnsureSubscribedToServerAsync(this, channel, CommandFlags.None, false, serverEndPoint)
+ : 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 7df8b8321..a3b6d9968 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,23 +558,27 @@ 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)
+ try
{
- switch (message.Command)
+ var ep = connection.BridgeCouldBeNull?.ServerEndPoint;
+ if (ep is not null)
{
- case RedisCommand.SUBSCRIBE:
- case RedisCommand.SSUBSCRIBE:
- case RedisCommand.PSUBSCRIBE:
- 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;
+ }
}
}
+ 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 767ab0a14..15025fcec 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);
@@ -77,7 +79,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 +111,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,129 +201,383 @@ 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;
+ private bool _needsRestore;
- 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 !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)
+ => IsConnectedAny() && _currentServer?.EndPoint == endpoint;
+
+ internal override 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)
{
- _currentServer = null;
+ if (_currentServer != expected) return false;
+ _needsRestore = true;
+ if (!IsConnectedAny())
+ {
+ _currentServer = null;
+ Volatile.Write(ref _currentConnection, null);
+ }
return true;
}
-
- return false;
}
- internal override bool UnsubscribeFromServer(
- RedisSubscriber subscriber,
- in RedisChannel channel,
- CommandFlags flags,
- bool internalCall)
+ internal ServerEndPoint? GetCurrentServer() => Volatile.Read(ref _currentServer);
+ internal override ServerEndPoint? GetAnyCurrentServer() => GetCurrentServer();
+
+ internal override void RemoveDisconnectedEndpoints()
{
- var server = _currentServer;
- if (server is not null)
+ RestoreOperation? superseded = null;
+ lock (_stateLock)
{
- var message = GetSubscriptionMessage(channel, SubscriptionAction.Unsubscribe, flags, internalCall);
- return subscriber.multiplexer.ExecuteSyncImpl(message, Processor, server);
+ if (_currentServer is not null && !IsConnectedAny())
+ {
+ _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();
+ }
- return true;
+ internal override int EnsureSubscribedToServer(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall)
+ {
+ RemoveDisconnectedEndpoints();
+ bool fireAndForget = ((Flags | flags) & CommandFlags.FireAndForget) != 0;
+ if (!Enter(subscriber, fireAndForget)) return 0;
+ RestoreOperation? operation = null;
+ bool transferred = false;
+ try
+ {
+ 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 Task UnsubscribeFromServerAsync(
- RedisSubscriber subscriber,
- RedisChannel channel,
- CommandFlags flags,
- object? asyncState,
- bool internalCall)
+ internal override async Task EnsureSubscribedToServerAsync(RedisSubscriber subscriber, RedisChannel channel, CommandFlags flags, bool internalCall, ServerEndPoint? server = null)
{
- var server = _currentServer;
- if (server is not null)
+ RemoveDisconnectedEndpoints();
+ bool fireAndForget = ((Flags | flags) & CommandFlags.FireAndForget) != 0;
+ if (fireAndForget)
{
- var message = GetSubscriptionMessage(channel, SubscriptionAction.Unsubscribe, flags, internalCall);
- return subscriber.multiplexer.ExecuteAsyncImpl(message, Processor, asyncState, server);
+ 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 AwaitRestoreAsync(operation).ForAwait();
+ return 1;
+ }
+ finally { if (!transferred) EndOperation(operation); }
+ }
- return CompletedTask.FromResult(true, asyncState);
+ private bool Enter(RedisSubscriber subscriber, bool fireAndForget)
+ {
+ if (_operationLock.Wait(fireAndForget ? 0 : subscriber.multiplexer.TimeoutMilliseconds)) return true;
+ if (fireAndForget) return false;
+ throw WaitingTimeout(subscriber);
}
- private ServerEndPoint? _currentServer;
- internal ServerEndPoint? GetCurrentServer() => Volatile.Read(ref _currentServer);
+ private Exception WaitingTimeout(RedisSubscriber subscriber)
+ => ExceptionFactory.Timeout(subscriber.multiplexer, "Timeout waiting for an in-flight subscription operation.", null, _currentServer);
- internal override ServerEndPoint? GetAnyCurrentServer() => Volatile.Read(ref _currentServer);
+ private RestoreOperation? BeginRestore(RedisSubscriber subscriber, RedisChannel channel, CommandFlags flags, bool internalCall, ServerEndPoint? server)
+ {
+ lock (_stateLock)
+ {
+ if (_unsubscribed) return 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);
+ return _pendingRestore = new RestoreOperation(this, subscriber, channel, message, server);
+ }
+ }
- ///
- /// Evaluates state and if we're not currently connected, clears the server reference.
- ///
- internal override void RemoveDisconnectedEndpoints()
+ private async Task CompleteRestoreAsync(RestoreOperation operation)
{
- var server = _currentServer;
- if (server is { IsSubscriberConnected: false })
+ 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)
{
- _currentServer = null;
+ _ = reply.ObserveErrors();
+ throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, CommandFlags.None, "The subscription restore was superseded by a live routing owner.");
}
+ await reply.ForAwait();
}
- internal override int EnsureSubscribedToServer(
- RedisSubscriber subscriber,
- in RedisChannel channel,
- CommandFlags flags,
- bool internalCall)
+ private void EndOperation(RestoreOperation? operation)
+ {
+ lock (_stateLock)
+ {
+ 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");
+ }
+ }
+
+ private void OnSubscribed(RestoreOperation operation, PhysicalConnection connection)
{
- RemoveIncorrectRouting(subscriber, in channel, flags, internalCall);
- if (IsConnectedAny()) return 0;
+ var server = connection.BridgeCouldBeNull?.ServerEndPoint;
+ if (server is null) return;
+ CleanupRequest? cleanup = null;
+ lock (_stateLock)
+ {
+ 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);
+ _needsRestore = false;
+ }
+ else
+ {
+ // An obsolete acknowledgement must not change the confirmed delivery route.
+ cleanup = new CleanupRequest(operation.Subscriber, operation.Channel, server);
+ }
+ }
+ if (cleanup is not null) QueueCleanup(cleanup);
+ }
- // 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 void QueueCleanup(CleanupRequest request)
+ {
+ lock (_stateLock)
+ {
+ (_cleanup ??= new()).Add(request);
+ }
+ DrainCleanup();
}
- private void RemoveIncorrectRouting(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall)
+ private void DrainCleanup()
{
- // 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)
+ // Serialize cleanup writes with restoration, but never write or forward under _stateLock.
+ if (!_operationLock.Wait(0)) return;
+ bool ownsOperation = true;
+ try
{
- // 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))
+ while (true)
{
- var message = GetSubscriptionMessage(channel, SubscriptionAction.Unsubscribe, flags | CommandFlags.FireAndForget, internalCall);
- subscriber.multiplexer.ExecuteSyncImpl(message, Processor, current);
- _currentServer = null; // pre-emptively disconnect - F+F
+ System.Collections.Generic.List? cleanup;
+ lock (_stateLock)
+ {
+ 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); }
+ }
}
}
+ finally { if (ownsOperation) _operationLock.Release(); }
}
- internal override async Task EnsureSubscribedToServerAsync(
- RedisSubscriber subscriber,
- RedisChannel channel,
- CommandFlags flags,
- bool internalCall,
- ServerEndPoint? server = null)
+ private void QueueUnsubscribeCleanup(RedisSubscriber subscriber, RedisChannel channel)
{
- RemoveIncorrectRouting(subscriber, in channel, flags, internalCall);
- if (IsConnectedAny()) return 0;
+ 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();
+ }
- // 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;
+ internal override bool UnsubscribeFromServer(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall)
+ {
+ Volatile.Write(ref _unsubscribed, true);
+ if (((Flags | flags) & CommandFlags.FireAndForget) != 0)
+ {
+ UnsubscribeCoreAsync(subscriber, channel, flags, null, internalCall).ObserveErrors();
+ return true;
+ }
+ if (!_operationLock.Wait(subscriber.multiplexer.TimeoutMilliseconds))
+ {
+ QueueUnsubscribeCleanup(subscriber, channel);
+ return true;
+ }
+ 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 | 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())
+ {
+ QueueUnsubscribeCleanup(subscriber, channel);
+ return true;
+ }
+ 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
+ {
+ 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;
+ 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; }
+ }
+
+ 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;
}
}
diff --git a/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs b/tests/StackExchange.Redis.Tests/InProcessSentinelServer.cs
index 8fecb98e6..8e98b7d41 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
new file mode 100644
index 000000000..2f5a444fb
--- /dev/null
+++ b/tests/StackExchange.Redis.Tests/SentinelSwitchCoalescingTests.cs
@@ -0,0 +1,126 @@
+using System;
+using System.Net;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+[RunPerProtocol]
+public class SentinelSwitchCoalescingTests(ITestOutputHelper output)
+{
+ [Fact]
+ 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);
+ 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;
+ 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()
+ {
+ 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();
+ }
+}
diff --git a/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs b/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs
new file mode 100644
index 000000000..9413a6e37
--- /dev/null
+++ b/tests/StackExchange.Redis.Tests/SubscriptionRestorationTests.cs
@@ -0,0 +1,413 @@
+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.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());
+ }
+ 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);
+ }
+
+ [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, 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();
+
+ 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 (!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.");
+ }
+ }
+ 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;
+ }
+ }
+}