feat(gossip): inject member and block list dependencies (#2357)

rogeralsing · web-flow · commit c2c6c1f2e8d9 · 2025-08-23T14:02:51.000+02:00
diff --git a/logs/log1755949362.md b/logs/log1755949362.md
@@ -0,0 +1,5 @@
+- Introduced IMemberList and IBlockList interfaces to decouple GossipActor from concrete MemberList and BlockList implementations.
+- Modified GossipActor to receive these interfaces, improving testability and making member/block lookups explicit dependencies.
+- Updated GossipSender to operate on IMemberList, reducing reliance on full Cluster object.
+- Adjusted Gossiper to supply MemberList and BlockList when creating GossipActor, ensuring compatibility with new constructor.
+- Revised GossipTransport tests to accommodate new GossipSender signature.
diff --git a/src/Proto.Cluster/Gossip/GossipActor.cs b/src/Proto.Cluster/Gossip/GossipActor.cs
@@ -20,6 +20,8 @@ public class GossipActor : IActor
 #pragma warning restore CS0618 // Type or member is obsolete
     private readonly TimeSpan _gossipRequestTimeout;
     private readonly IGossip _internal;
+    private readonly IMemberList _memberList;
+    private readonly IBlockList _blockList;
     private readonly IGossipTransport _transport;
 
     // lookup from state key -> consensus checks
@@ -31,12 +33,16 @@ public GossipActor(
         int gossipFanout,
         int gossipMaxSend,
         IGossip gossip,
-        IGossipTransport transport
+        IGossipTransport transport,
+        IMemberList memberList,
+        IBlockList blockList
     )
     {
         _gossipRequestTimeout = gossipRequestTimeout;
         _internal = gossip;
         _transport = transport;
+        _memberList = memberList;
+        _blockList = blockList;
     }
 
     public async Task ReceiveAsync(IContext context)
@@ -121,7 +127,7 @@ private Task OnGossipRequest(IContext context, GossipRequest gossipRequest)
         var logger = context.Logger()?.BeginScope<GossipActor>();
         logger?.LogDebug("Gossip Request {Sender}", context.Sender!);
         
-        if (context.Remote().BlockList.BlockedMembers.Contains(gossipRequest.MemberId))
+        if (_blockList.BlockedMembers.Contains(gossipRequest.MemberId))
         {
             Logger.LogInformation("Blocked gossip request from {MemberId}", gossipRequest.MemberId);
             context.Respond(new GossipResponse()
@@ -130,8 +136,7 @@ private Task OnGossipRequest(IContext context, GossipRequest gossipRequest)
             });
             return Task.CompletedTask;
         }
-
-        if (!context.Cluster().MemberList.ContainsMemberId(gossipRequest.MemberId))
+        if (!_memberList.ContainsMemberId(gossipRequest.MemberId))
         {
             Logger.LogInformation("Ignoring gossip request from {MemberId} as it is not a member", gossipRequest.MemberId);
             context.Respond(new GossipResponse()
@@ -140,18 +145,13 @@ private Task OnGossipRequest(IContext context, GossipRequest gossipRequest)
             });
             return Task.CompletedTask;
         }
-        
-        
-        
 
         if (Logger.IsEnabled(LogLevel.Debug))
         {
             Logger.LogDebug("Gossip Request {Sender}", context.Sender!);
         }
         
         ReceiveState(context, gossipRequest.State);
-
-        
         if (context.Cluster().Config.GossipDebugLogging)
         {
             Logger.LogInformation("Responding to GossipRequest {Request} to {MemberId}", gossipRequest, gossipRequest.MemberId);
@@ -216,6 +216,6 @@ private void SendGossipForMember(IContext context, Member targetMember,
             Logger.LogInformation("Sending GossipRequest {Request} to {MemberId}", gossipRequest, targetMember.Id);
         }
 
-        GossipSender.Send(context, context.Cluster(), targetMember, memberStateDelta, gossipRequest, _gossipRequestTimeout, _transport);
+        GossipSender.Send(context, _memberList, targetMember, memberStateDelta, gossipRequest, _gossipRequestTimeout, _transport);
     }
 }
diff --git a/src/Proto.Cluster/Gossip/GossipSender.cs b/src/Proto.Cluster/Gossip/GossipSender.cs
@@ -10,6 +10,7 @@
 using Microsoft.Extensions.Logging;
 using Proto;
 using Proto.Logging;
+using Proto.Cluster;
 
 namespace Proto.Cluster.Gossip;
 
@@ -19,7 +20,7 @@ internal static class GossipSender
 
     public static void Send(
         IContext context,
-        Cluster cluster,
+        IMemberList memberList,
         Member targetMember,
         MemberStateDelta memberStateDelta,
         GossipRequest request,
@@ -36,9 +37,9 @@ public static void Send(
             async task =>
             {
                 var delta = DateTime.UtcNow - start;
-                var self = cluster.MemberList.Self;
+                var self = memberList.Self;
 
-                if (!cluster.MemberList.TryGetMember(targetMember.Id, out _))
+                if (!memberList.TryGetMember(targetMember.Id, out _))
                 {
                     return;
                 }
@@ -72,7 +73,7 @@ public static void Send(
                 }
                 catch (Exception x)
                 {
-                    if (cluster.MemberList.TryGetMember(targetMember.Id, out _))
+                    if (memberList.TryGetMember(targetMember.Id, out _))
                     {
                         Logger.LogError(
                             x,
diff --git a/src/Proto.Cluster/Gossip/Gossiper.cs b/src/Proto.Cluster/Gossip/Gossiper.cs
@@ -204,7 +204,9 @@ internal Task StartGossipActorAsync(IGossip? gossip = null, IGossipTransport? tr
             _cluster.Config.GossipFanout,
             _cluster.Config.GossipMaxSend,
             _gossip,
-            transport ?? new GossipTransport()));
+            transport ?? new GossipTransport(),
+            _cluster.MemberList,
+            _cluster.System.Remote().BlockList));
 
         _pid = _context.SpawnNamedSystem(props, GossipActorName);
         _cluster.System.EventStream.Subscribe<ClusterTopology>(topology =>
diff --git a/src/Proto.Cluster/Membership/IMemberList.cs b/src/Proto.Cluster/Membership/IMemberList.cs
@@ -0,0 +1,9 @@
+namespace Proto.Cluster;
+
+public interface IMemberList
+{
+    bool ContainsMemberId(string memberId);
+    bool TryGetMember(string memberId, out Member? value);
+    Member Self { get; }
+}
+
diff --git a/src/Proto.Cluster/Membership/MemberList.cs b/src/Proto.Cluster/Membership/MemberList.cs
@@ -27,7 +27,7 @@ namespace Proto.Cluster;
 ///     If the member learns that it is blocked from gossip, it will initiate shutdown.
 /// </summary>
 [PublicAPI]
-public record MemberList
+public record MemberList : IMemberList
 {
 #pragma warning disable CS0618 // Type or member is obsolete
     private static readonly ILogger Logger = Log.CreateLogger<MemberList>();
diff --git a/src/Proto.Remote/BlockList.cs b/src/Proto.Remote/BlockList.cs
@@ -18,7 +18,7 @@ public record MemberBlocked(string MemberId, string Reason);
 ///     unresponsiveness. Entries on this list expire after <see cref="ActorSystemConfig.BlockedMemberDuration"/>,
 ///     defaults to 1 hour.
 /// </summary>
-public class BlockList
+public class BlockList : IBlockList
 {
     private readonly object _lock = new();
     private readonly ActorSystem _system;
diff --git a/src/Proto.Remote/IBlockList.cs b/src/Proto.Remote/IBlockList.cs
@@ -0,0 +1,9 @@
+using System.Collections.Immutable;
+
+namespace Proto.Remote;
+
+public interface IBlockList
+{
+    ImmutableHashSet<string> BlockedMembers { get; }
+}
+
diff --git a/tests/Proto.Cluster.Tests/GossipTransportTests.cs b/tests/Proto.Cluster.Tests/GossipTransportTests.cs
@@ -54,7 +54,7 @@ public async Task Success_should_commit_offsets()
 
         var props = Props.FromFunc(ctx =>
         {
-            GossipSender.Send(ctx, cluster, targetMember, delta, request, cluster.Config.GossipRequestTimeout, transport);
+            GossipSender.Send(ctx, cluster.MemberList, targetMember, delta, request, cluster.Config.GossipRequestTimeout, transport);
             return Task.CompletedTask;
         });
 
@@ -87,7 +87,7 @@ public async Task Rejection_should_not_commit_offsets()
 
         var props = Props.FromFunc(ctx =>
         {
-            GossipSender.Send(ctx, cluster, targetMember, delta, request, cluster.Config.GossipRequestTimeout, transport);
+            GossipSender.Send(ctx, cluster.MemberList, targetMember, delta, request, cluster.Config.GossipRequestTimeout, transport);
             return Task.CompletedTask;
         });
 
@@ -119,7 +119,7 @@ public async Task Timeout_should_not_commit_offsets()
 
         var props = Props.FromFunc(ctx =>
         {
-            GossipSender.Send(ctx, cluster, targetMember, delta, request, cluster.Config.GossipRequestTimeout, transport);
+            GossipSender.Send(ctx, cluster.MemberList, targetMember, delta, request, cluster.Config.GossipRequestTimeout, transport);
             return Task.CompletedTask;
         });
 
@@ -151,7 +151,7 @@ public async Task DeadLetter_should_not_commit_offsets()
 
         var props = Props.FromFunc(ctx =>
         {
-            GossipSender.Send(ctx, cluster, targetMember, delta, request, cluster.Config.GossipRequestTimeout, transport);
+            GossipSender.Send(ctx, cluster.MemberList, targetMember, delta, request, cluster.Config.GossipRequestTimeout, transport);
             return Task.CompletedTask;
         });
 

Original file line number	Diff line number	Diff line change
`@@ -20,6 +20,8 @@ public class GossipActor : IActor`
`20`	`20`	`#pragma warning restore CS0618 // Type or member is obsolete`
`21`	`21`	`private readonly TimeSpan _gossipRequestTimeout;`
`22`	`22`	`private readonly IGossip _internal;`
	`23`	`+ private readonly IMemberList _memberList;`
	`24`	`+ private readonly IBlockList _blockList;`
`23`	`25`	`private readonly IGossipTransport _transport;`
`24`	`26`
`25`	`27`	`// lookup from state key -> consensus checks`
`@@ -31,12 +33,16 @@ public GossipActor(`
`31`	`33`	`int gossipFanout,`
`32`	`34`	`int gossipMaxSend,`
`33`	`35`	`IGossip gossip,`
`34`		`- IGossipTransport transport`
	`36`	`+ IGossipTransport transport,`
	`37`	`+ IMemberList memberList,`
	`38`	`+ IBlockList blockList`
`35`	`39`	`)`
`36`	`40`	`{`
`37`	`41`	`_gossipRequestTimeout = gossipRequestTimeout;`
`38`	`42`	`_internal = gossip;`
`39`	`43`	`_transport = transport;`
	`44`	`+ _memberList = memberList;`
	`45`	`+ _blockList = blockList;`
`40`	`46`	`}`
`41`	`47`
`42`	`48`	`public async Task ReceiveAsync(IContext context)`
`@@ -121,7 +127,7 @@ private Task OnGossipRequest(IContext context, GossipRequest gossipRequest)`
`121`	`127`	`var logger = context.Logger()?.BeginScope<GossipActor>();`
`122`	`128`	`logger?.LogDebug("Gossip Request {Sender}", context.Sender!);`
`123`	`129`
`124`		`- if (context.Remote().BlockList.BlockedMembers.Contains(gossipRequest.MemberId))`
	`130`	`+ if (_blockList.BlockedMembers.Contains(gossipRequest.MemberId))`
`125`	`131`	`{`
`126`	`132`	`Logger.LogInformation("Blocked gossip request from {MemberId}", gossipRequest.MemberId);`
`127`	`133`	`context.Respond(new GossipResponse()`
`@@ -130,8 +136,7 @@ private Task OnGossipRequest(IContext context, GossipRequest gossipRequest)`
`130`	`136`	`});`
`131`	`137`	`return Task.CompletedTask;`
`132`	`138`	`}`
`133`		`-`
`134`		`- if (!context.Cluster().MemberList.ContainsMemberId(gossipRequest.MemberId))`
	`139`	`+ if (!_memberList.ContainsMemberId(gossipRequest.MemberId))`
`135`	`140`	`{`
`136`	`141`	`Logger.LogInformation("Ignoring gossip request from {MemberId} as it is not a member", gossipRequest.MemberId);`
`137`	`142`	`context.Respond(new GossipResponse()`
`@@ -140,18 +145,13 @@ private Task OnGossipRequest(IContext context, GossipRequest gossipRequest)`
`140`	`145`	`});`
`141`	`146`	`return Task.CompletedTask;`
`142`	`147`	`}`
`143`		`-`
`144`		`-`
`145`		`-`
`146`	`148`
`147`	`149`	`if (Logger.IsEnabled(LogLevel.Debug))`
`148`	`150`	`{`
`149`	`151`	`Logger.LogDebug("Gossip Request {Sender}", context.Sender!);`
`150`	`152`	`}`
`151`	`153`
`152`	`154`	`ReceiveState(context, gossipRequest.State);`
`153`		`-`
`154`		`-`
`155`	`155`	`if (context.Cluster().Config.GossipDebugLogging)`
`156`	`156`	`{`
`157`	`157`	`Logger.LogInformation("Responding to GossipRequest {Request} to {MemberId}", gossipRequest, gossipRequest.MemberId);`
`@@ -216,6 +216,6 @@ private void SendGossipForMember(IContext context, Member targetMember,`
`216`	`216`	`Logger.LogInformation("Sending GossipRequest {Request} to {MemberId}", gossipRequest, targetMember.Id);`
`217`	`217`	`}`
`218`	`218`
`219`		`- GossipSender.Send(context, context.Cluster(), targetMember, memberStateDelta, gossipRequest, _gossipRequestTimeout, _transport);`
	`219`	`+ GossipSender.Send(context, _memberList, targetMember, memberStateDelta, gossipRequest, _gossipRequestTimeout, _transport);`
`220`	`220`	`}`
`221`	`221`	`}`
Original file line number	Diff line number	Diff line change
`@@ -10,6 +10,7 @@`
`10`	`10`	`using Microsoft.Extensions.Logging;`
`11`	`11`	`using Proto;`
`12`	`12`	`using Proto.Logging;`
	`13`	`+using Proto.Cluster;`
`13`	`14`
`14`	`15`	`namespace Proto.Cluster.Gossip;`
`15`	`16`
`@@ -19,7 +20,7 @@ internal static class GossipSender`
`19`	`20`
`20`	`21`	`public static void Send(`
`21`	`22`	`IContext context,`
`22`		`- Cluster cluster,`
	`23`	`+ IMemberList memberList,`
`23`	`24`	`Member targetMember,`
`24`	`25`	`MemberStateDelta memberStateDelta,`
`25`	`26`	`GossipRequest request,`
`@@ -36,9 +37,9 @@ public static void Send(`
`36`	`37`	`async task =>`
`37`	`38`	`{`
`38`	`39`	`var delta = DateTime.UtcNow - start;`
`39`		`- var self = cluster.MemberList.Self;`
	`40`	`+ var self = memberList.Self;`
`40`	`41`
`41`		`- if (!cluster.MemberList.TryGetMember(targetMember.Id, out _))`
	`42`	`+ if (!memberList.TryGetMember(targetMember.Id, out _))`
`42`	`43`	`{`
`43`	`44`	`return;`
`44`	`45`	`}`
`@@ -72,7 +73,7 @@ public static void Send(`
`72`	`73`	`}`
`73`	`74`	`catch (Exception x)`
`74`	`75`	`{`
`75`		`- if (cluster.MemberList.TryGetMember(targetMember.Id, out _))`
	`76`	`+ if (memberList.TryGetMember(targetMember.Id, out _))`
`76`	`77`	`{`
`77`	`78`	`Logger.LogError(`
`78`	`79`	`x,`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ namespace Proto.Cluster;`
`27`	`27`	`/// If the member learns that it is blocked from gossip, it will initiate shutdown.`
`28`	`28`	`/// </summary>`
`29`	`29`	`[PublicAPI]`
`30`		`-public record MemberList`
	`30`	`+public record MemberList : IMemberList`
`31`	`31`	`{`
`32`	`32`	`#pragma warning disable CS0618 // Type or member is obsolete`
`33`	`33`	`private static readonly ILogger Logger = Log.CreateLogger<MemberList>();`
Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,7 @@ public record MemberBlocked(string MemberId, string Reason);`
`18`	`18`	`/// unresponsiveness. Entries on this list expire after <see cref="ActorSystemConfig.BlockedMemberDuration"/>,`
`19`	`19`	`/// defaults to 1 hour.`
`20`	`20`	`/// </summary>`
`21`		`-public class BlockList`
	`21`	`+public class BlockList : IBlockList`
`22`	`22`	`{`
`23`	`23`	`private readonly object _lock = new();`
`24`	`24`	`private readonly ActorSystem _system;`