Let event subscribers attribute group-forwarded events to their member

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.

src/StackExchange.Redis/Availability/ConnectionGroupExtensions.cs

@@ -0,0 +1,25 @@
+namespace StackExchange.Redis.Availability;
+
+/// <summary>
+/// Convenience methods for working with an <see cref="IConnectionGroup"/>.
+/// </summary>
+public static class ConnectionGroupExtensions
+{
+    /// <summary>
+    /// Find the member of the group that the sender of an event originated from, if any.
+    /// </summary>
+    /// <param name="group">The group that raised the event.</param>
+    /// <param name="sender">The <c>sender</c> passed to an event handler.</param>
+    /// <returns>The member that the event came from, or <c>null</c> 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.</returns>
+    /// <remarks>See <see cref="ConnectionGroupMember.IsSenderOf"/>.</remarks>
+    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;
+    }
+}

src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs

@@ -134,6 +134,17 @@ internal void SetUnhealthy()
             /// </summary>
             public bool IsUnhealthy => GetFlag(MemberFlags.Unhealthy);
 
+            /// <summary>
+            /// 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 <see cref="IConnectionMultiplexer.ServerMaintenanceEvent"/>
+            /// or <see cref="IConnectionMultiplexer.ConnectionFailed"/>) are raised with the member's own connection as the sender, so this
+            /// identifies which member an event came from; see also <see cref="ConnectionGroupExtensions.FindMember"/>.
+            /// </summary>
+            /// <param name="sender">The <c>sender</c> passed to an event handler.</param>
+            /// <remarks>This is <c>false</c> for the group itself, and for a member that has been removed from its group.</remarks>
+            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;
                         }

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?

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+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<object?>();
+        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));
+    }
+}