From 9bef87bb3fbb8da970c326568de1e0c4bc091a28 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Fri, 2 Oct 2026 15:46:57 +0100 Subject: [PATCH] Let event subscribers attribute group-forwarded events to their member Adds ConnectionGroupMember.IsSenderOf(object?) and ConnectionGroupExtensions.FindMember, so that a handler for an event forwarded by a connection group (ServerMaintenanceEvent, ConnectionFailed, ...) can say which member it came from, without exposing the member's connection. See #3249. --- docs/Failover.md | 21 ++++ .../Availability/ConnectionGroupExtensions.cs | 25 +++++ .../Availability/MultiGroupMultiplexer.cs | 13 ++- .../PublicAPI/PublicAPI.Unshipped.txt | 3 + .../GroupMemberSenderUnitTests.cs | 99 +++++++++++++++++++ 5 files changed, 160 insertions(+), 1 deletion(-) create mode 100644 src/StackExchange.Redis/Availability/ConnectionGroupExtensions.cs create mode 100644 tests/StackExchange.Redis.Tests/GroupMemberSenderUnitTests.cs diff --git a/docs/Failover.md b/docs/Failover.md index 98d717a90f..44ffc7e7b7 100644 --- a/docs/Failover.md +++ b/docs/Failover.md @@ -281,6 +281,27 @@ foreach (var member in conn.GetMembers()) These are the same instances that were passed into `ConnectGroupAsync`. +## Which member raised an event + +Events the group forwards from its members (`ConnectionFailed`, `ConnectionRestored`, `ErrorMessage`, `ServerMaintenanceEvent`, +`HashSlotMoved`, and so on) arrive with the member's own connection as `sender`, not the group. To say which member an event +came from, use `FindMember`, or ask a specific member with `IsSenderOf`: + +```csharp +conn.ConnectionFailed += (sender, args) => +{ + var member = conn.FindMember(sender); // null if the sender is not (or is no longer) a member + Console.WriteLine($"{member?.Name ?? "(unknown)"}: {args.FailureType}"); + + if (conn.ActiveMember?.IsSenderOf(sender) == true) + { + Console.WriteLine("...and that is the member currently in use"); + } +}; +``` + +Bear in mind that the active member can change at any time, including just after this check. + ## Health Checks Configurable health checking monitors the health of all endpoints and automatically routes traffic away from unhealthy instances. diff --git a/src/StackExchange.Redis/Availability/ConnectionGroupExtensions.cs b/src/StackExchange.Redis/Availability/ConnectionGroupExtensions.cs new file mode 100644 index 0000000000..29350b039f --- /dev/null +++ b/src/StackExchange.Redis/Availability/ConnectionGroupExtensions.cs @@ -0,0 +1,25 @@ +namespace StackExchange.Redis.Availability; + +/// +/// Convenience methods for working with an . +/// +public static class ConnectionGroupExtensions +{ + /// + /// Find the member of the group that the sender of an event originated from, if any. + /// + /// The group that raised the event. + /// The sender passed to an event handler. + /// The member that the event came from, or null if the sender is not (or is no longer) one of this group's members, + /// including when the event was raised by the group itself. + /// See . + public static ConnectionGroupMember? FindMember(this IConnectionGroup group, object? sender) + { + if (group is null || sender is null) return null; + foreach (var member in group.GetMembers()) + { + if (member.IsSenderOf(sender)) return member; + } + return null; + } +} diff --git a/src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs b/src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs index f4bf36872b..448faa3cfd 100644 --- a/src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs +++ b/src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs @@ -134,6 +134,17 @@ internal void SetUnhealthy() /// public bool IsUnhealthy => GetFlag(MemberFlags.Unhealthy); + /// + /// Whether the sender of an event originated from this member: either this member itself, or the connection it currently + /// holds. Events that a group forwards from its members (for example + /// or ) are raised with the member's own connection as the sender, so this + /// identifies which member an event came from; see also . + /// + /// The sender passed to an event handler. + /// This is false for the group itself, and for a member that has been removed from its group. + public bool IsSenderOf(object? sender) + => sender is not null && (ReferenceEquals(sender, this) || ReferenceEquals(sender, Volatile.Read(ref _muxer))); + [DoesNotReturn] private static ConnectionMultiplexer ThrowNoMuxer() => throw new InvalidOperationException("Member is not connected."); @@ -424,7 +435,7 @@ public override string ToString() { foreach (var member in _members) { - if (ReferenceEquals(muxer, member.Multiplexer)) + if (member.IsSenderOf(muxer)) { return member; } diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index ab058de62d..ce6a048bb1 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -1 +1,4 @@ #nullable enable +StackExchange.Redis.Availability.ConnectionGroupExtensions +StackExchange.Redis.Availability.ConnectionGroupMember.IsSenderOf(object? sender) -> bool +static StackExchange.Redis.Availability.ConnectionGroupExtensions.FindMember(this StackExchange.Redis.Availability.IConnectionGroup! group, object? sender) -> StackExchange.Redis.Availability.ConnectionGroupMember? diff --git a/tests/StackExchange.Redis.Tests/GroupMemberSenderUnitTests.cs b/tests/StackExchange.Redis.Tests/GroupMemberSenderUnitTests.cs new file mode 100644 index 0000000000..5506b2ce1c --- /dev/null +++ b/tests/StackExchange.Redis.Tests/GroupMemberSenderUnitTests.cs @@ -0,0 +1,99 @@ +using System; +using System.Collections.Concurrent; +using System.Threading.Tasks; +using StackExchange.Redis.Availability; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Events that a group forwards from its members arrive with the member's own connection as the sender; a subscriber +/// must be able to say which member that was, without the connection itself being part of the public surface. +/// +public class GroupMemberSenderUnitTests(ITestOutputHelper log) +{ + [Fact] + public async Task ForwardedEventsAreAttributedToTheirMember() + { + using var server0 = new InProcessTestServer(log); + using var server1 = new InProcessTestServer(log); + var member0 = new ConnectionGroupMember(server0.GetClientConfig(defaultOnly: true), "zero"); + var member1 = new ConnectionGroupMember(server1.GetClientConfig(defaultOnly: true), "one"); + + var group = await ConnectionMultiplexer.ConnectGroupAsync(member0, member1); + await using var _ = (IAsyncDisposable)group; + + var senders = new ConcurrentQueue(); + group.ServerMaintenanceEvent += (sender, _) => senders.Enqueue(sender); + + // raised exactly as a real notification would be: on the member's own connection, then forwarded by the group + member0.Multiplexer.OnServerMaintenanceEvent(new AzureMaintenanceEvent("NotificationType|NodeMaintenanceStarting")); + member1.Multiplexer.OnServerMaintenanceEvent(new AzureMaintenanceEvent("NotificationType|NodeMaintenanceEnded")); + + Assert.Equal(2, senders.Count); + Assert.True(senders.TryDequeue(out var first)); + Assert.True(senders.TryDequeue(out var second)); + + Assert.True(member0.IsSenderOf(first)); + Assert.False(member1.IsSenderOf(first)); + Assert.Same(member0, group.FindMember(first)); + + Assert.True(member1.IsSenderOf(second)); + Assert.False(member0.IsSenderOf(second)); + Assert.Same(member1, group.FindMember(second)); + + // "which member is this from" is a different question to "is this about what I am using right now", but + // the latter is now expressible + var active = group.ActiveMember; + Assert.NotNull(active); + Assert.True(active.IsSenderOf(first) ^ active.IsSenderOf(second)); + } + + [Fact] + public async Task NonMemberSendersAreNotClaimed() + { + using var server0 = new InProcessTestServer(log); + using var server1 = new InProcessTestServer(log); + var member0 = new ConnectionGroupMember(server0.GetClientConfig(defaultOnly: true), "zero"); + var member1 = new ConnectionGroupMember(server1.GetClientConfig(defaultOnly: true), "one"); + + var group = await ConnectionMultiplexer.ConnectGroupAsync(member0, member1); + await using var _ = (IAsyncDisposable)group; + + Assert.False(member0.IsSenderOf(null)); + Assert.Null(group.FindMember(null)); + + // the group itself is not any one member, even the active one + Assert.False(member0.IsSenderOf(group)); + Assert.False(member1.IsSenderOf(group)); + Assert.Null(group.FindMember(group)); + + Assert.False(member0.IsSenderOf(new object())); + + // a member is the sender of anything it raises itself + Assert.True(member0.IsSenderOf(member0)); + Assert.False(member0.IsSenderOf(member1)); + } + + [Fact] + public async Task RemovedMemberNoLongerClaimsItsConnection() + { + using var server0 = new InProcessTestServer(log); + using var server1 = new InProcessTestServer(log); + var member0 = new ConnectionGroupMember(server0.GetClientConfig(defaultOnly: true), "zero"); + var member1 = new ConnectionGroupMember(server1.GetClientConfig(defaultOnly: true), "one"); + + var group = await ConnectionMultiplexer.ConnectGroupAsync(member0, member1); + await using var _ = (IAsyncDisposable)group; + + var connection = member1.Multiplexer; + Assert.True(member1.IsSenderOf(connection)); + Assert.Same(member1, group.FindMember(connection)); + + Assert.True(group.Remove(member1)); + + Assert.False(member1.IsSenderOf(connection)); + Assert.Null(group.FindMember(connection)); + } +}