Skip to content

Pub/sub: overlapping subscription restores cannot register on two servers #3353

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mgravell wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Pub/sub: overlapping subscription restores cannot register on two servers #3353

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion src/StackExchange.Redis/ResultProcessor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -565,7 +565,24 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
case RedisCommand.SUBSCRIBE:
case RedisCommand.SSUBSCRIBE:
case RedisCommand.PSUBSCRIBE:
Subscription?.AddEndpoint(ep);
if (Subscription is { } sub && !sub.TryAdoptEndpoint(ep))
{
// the subscription is already confirmed on a different live server:
// this ack is a stray from an overlapping restore (#3336); remove
// the server-side registration it created, or it would deliver
// duplicate messages indefinitely (fire-and-forget: we are on the
// read loop, so we must not block)
if (message is Message.CommandChannelBase channelMessage)
{
var unsubscribe = sub.GetSubscriptionMessage(
channelMessage.Channel,
ConnectionMultiplexer.SubscriptionAction.Unsubscribe,
CommandFlags.FireAndForget,
internalCall: true);
connection.BridgeCouldBeNull?.Multiplexer.ExecuteSyncImpl(unsubscribe, this, ep);
}
}

break;
default:
Subscription?.TryRemoveEndpoint(ep);
Expand Down
172 changes: 161 additions & 11 deletions src/StackExchange.Redis/Subscription.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,10 @@ internal abstract class Subscription
internal abstract bool IsConnectedAny();
internal abstract bool IsConnectedTo(EndPoint endpoint);

internal abstract void AddEndpoint(ServerEndPoint server);
// Called when a (P|S)SUBSCRIBE acknowledgement arrives; returns false when the ack is a
// stray (we are already confirmed on a different, connected server) and the caller should
// remove the server-side registration it represents.
internal abstract bool TryAdoptEndpoint(ServerEndPoint server);

// conditional clear
internal abstract bool TryRemoveEndpoint(ServerEndPoint expected);
Expand Down Expand Up @@ -77,7 +80,8 @@ internal Message GetSubscriptionMessage(
in RedisChannel channel,
SubscriptionAction action,
CommandFlags flags,
bool internalCall)
bool internalCall,
bool stripFireAndForget = false)
{
const RedisChannel.RedisChannelOptions OPTIONS_MASK = ~(
RedisChannel.RedisChannelOptions.KeyRouted | RedisChannel.RedisChannelOptions.IgnoreChannelPrefix);
Expand Down Expand Up @@ -108,7 +112,17 @@ internal Message GetSubscriptionMessage(
};

// TODO: Consider flags here - we need to pass Fire and Forget, but don't want to intermingle Primary/Replica
var msg = Message.Create(-1, Flags | flags, command, channel);
var effectiveFlags = Flags | flags;
if (stripFireAndForget)
{
// a fire-and-forget message has no result box, so nothing observes its completion;
// the subscribe-claim logic needs that signal (ack, error reply, connection failure
// or timeout) to release a pending claim, so the message itself must not be
// fire-and-forget even when the caller is (the caller then simply doesn't await it)
effectiveFlags &= ~CommandFlags.FireAndForget;
}

var msg = Message.Create(-1, effectiveFlags, command, channel);
msg.SetForSubscriptionBridge();
if (internalCall)
{
Expand Down Expand Up @@ -206,7 +220,30 @@ internal override bool IsConnectedTo(EndPoint endpoint)
return server is { IsSubscriberConnected: true } && server.EndPoint == endpoint;
}

internal override void AddEndpoint(ServerEndPoint server) => _currentServer = server;
internal override bool TryAdoptEndpoint(ServerEndPoint server)
{
// an ack from the pending claim's server resolves that claim, whether or not we adopt it
var pending = Volatile.Read(ref _pending);
if (pending is not null && pending.Server == server)
{
Interlocked.CompareExchange(ref _pending, null, pending);
}

var current = Volatile.Read(ref _currentServer);
while (true)
{
if (current is { IsSubscriberConnected: true } && current != server)
{
// already confirmed on a different live server; the incoming ack is a stray
// from an overlapping restore, and should be unsubscribed by the caller
return false;
}

var seen = Interlocked.CompareExchange(ref _currentServer, server, current);
if (seen == current) return true;
current = seen;
}
}

internal override bool TryRemoveEndpoint(ServerEndPoint expected)
{
Expand Down Expand Up @@ -253,6 +290,12 @@ internal override Task<bool> UnsubscribeFromServerAsync(
}

private ServerEndPoint? _currentServer;

// an in-flight subscribe attempt; while one is pending, other restore attempts back off,
// so that overlapping restores (reconnect + reconfigure racing during a failover) cannot
// register the same subscription on two different servers (#3336)
private PendingClaim? _pending;

internal ServerEndPoint? GetCurrentServer() => Volatile.Read(ref _currentServer);

internal override ServerEndPoint? GetAnyCurrentServer() => Volatile.Read(ref _currentServer);
Expand All @@ -267,6 +310,12 @@ internal override void RemoveDisconnectedEndpoints()
{
_currentServer = null;
}

var pending = Volatile.Read(ref _pending);
if (pending is not null && !pending.Server.IsSubscriberConnected)
{
Interlocked.CompareExchange(ref _pending, null, pending);
}
}

internal override int EnsureSubscribedToServer(
Expand All @@ -278,14 +327,77 @@ internal override int EnsureSubscribedToServer(
RemoveIncorrectRouting(subscriber, in channel, flags, internalCall);
if (IsConnectedAny()) return 0;

bool callerIsFireAndForget = ((Flags | flags) & CommandFlags.FireAndForget) != 0;
var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall, stripFireAndForget: true);
var selected = subscriber.multiplexer.SelectServer(message);
if (!TryClaim(selected, out var claim)) return 0;

// we're not appropriately connected, so blank it out for eligible reconnection
_currentServer = null;
var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall);
var selected = subscriber.multiplexer.SelectServer(message);
_ = subscriber.ExecuteSync(message, Processor, selected);
if (callerIsFireAndForget)
{
try
{
// fire-and-observe: never block (restore paths run inside result processing),
// but watch the outcome so a failed attempt releases the claim for retry
ObserveClaim(subscriber.ExecuteAsync(message, Processor, selected), claim);
}
catch
{
Release(claim);
throw;
}
}
else
{
try
{
_ = subscriber.ExecuteSync(message, Processor, selected);
}
finally
{
// normally already resolved by the ack; backstop for failures
Release(claim);
}
}

return 1;
}

private bool TryClaim(ServerEndPoint? selected, out PendingClaim? claim)
{
if (selected is null)
{
// nothing selectable right now; attempt unclaimed (ack-time reconciliation
// still protects against duplicates)
claim = null;
return true;
}

claim = new PendingClaim(this, selected);
// if another subscribe attempt is already in flight, back off; its completion
// (ack or failure) will resolve the state, retried via the usual restore paths
return Interlocked.CompareExchange(ref _pending, claim, null) is null;
}

private void Release(PendingClaim? claim)
{
if (claim is not null)
{
Interlocked.CompareExchange(ref _pending, null, claim);
}
}

private static void ObserveClaim(Task task, PendingClaim? claim)
=> task.ContinueWith(
static (t, state) =>
{
_ = t.Exception; // observed; the retry policy is simply "the next restore attempt"
(state as PendingClaim)?.Release();
},
claim,
TaskContinuationOptions.ExecuteSynchronously);

private void RemoveIncorrectRouting(RedisSubscriber subscriber, in RedisChannel channel, CommandFlags flags, bool internalCall)
{
// only applies to cluster, when using key-routed channels (sharded, explicit key-routed, or
Expand Down Expand Up @@ -313,13 +425,48 @@ internal override async Task<int> EnsureSubscribedToServerAsync(
RemoveIncorrectRouting(subscriber, in channel, flags, internalCall);
if (IsConnectedAny()) return 0;

bool callerIsFireAndForget = ((Flags | flags) & CommandFlags.FireAndForget) != 0;
var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall, stripFireAndForget: true);
server ??= subscriber.multiplexer.SelectServer(message);
if (!TryClaim(server, out var claim)) return 0;

// we're not appropriately connected, so blank it out for eligible reconnection
_currentServer = null;
var message = GetSubscriptionMessage(channel, SubscriptionAction.Subscribe, flags, internalCall);
server ??= subscriber.multiplexer.SelectServer(message);
await subscriber.ExecuteAsync(message, Processor, server).ForAwait();
if (callerIsFireAndForget)
{
try
{
// fire-and-observe; see EnsureSubscribedToServer
ObserveClaim(subscriber.ExecuteAsync(message, Processor, server), claim);
}
catch
{
Release(claim);
throw;
}
}
else
{
try
{
await subscriber.ExecuteAsync(message, Processor, server).ForAwait();
}
finally
{
// normally already resolved by the ack; backstop for failures
Release(claim);
}
}

return 1;
}

private sealed class PendingClaim(SingleNodeSubscription owner, ServerEndPoint server)
{
public ServerEndPoint Server { get; } = server;

public void Release() => owner.Release(this);
}
}

// used for keyspace subscriptions, which are routed to multiple nodes
Expand Down Expand Up @@ -352,13 +499,16 @@ internal override bool IsConnectedTo(EndPoint endpoint)
=> _servers.TryGetValue(endpoint, out var server)
&& server.IsSubscriberConnected;

internal override void AddEndpoint(ServerEndPoint server)
internal override bool TryAdoptEndpoint(ServerEndPoint server)
{
// multi-node subscriptions want every eligible node, so an ack is never a stray
var ep = server.EndPoint;
if (!_servers.TryAdd(ep, server))
{
_servers[ep] = server;
}

return true;
}

internal override bool TryRemoveEndpoint(ServerEndPoint expected)
Expand Down
Loading
Loading