dohertj2/natsdotnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add RaftGroup lifecycle methods (Gap 2.9)

Browse Source 
Implement IsMember, SetPreferred, RemovePeer, AddPeer, CreateRaftGroup
factory, IsUnderReplicated, and IsOverReplicated on RaftGroup. Add 22
RaftGroupLifecycleTests covering all helpers and quorum size calculation.
This commit is contained in:
Joseph Doherty
2026-02-25 08:59:36 -05:00
parent 38ae1f6bea
commit f69f9b3220
2 changed files with 334 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
src/NATS.Server/JetStream/Cluster/ClusterAssignmentTypes.cs
View File

@@ -28,6 +28,80 @@ public sealed class RaftGroup
public int QuorumSize => (Peers.Count / 2) + 1;
public bool HasQuorum(int ackCount) => ackCount >= QuorumSize;
/// <summary>
/// True when the group has fewer peers than the desired replica count.
/// Go reference: jetstream_cluster.go missingPeers — len(rg.Peers) &lt; DesiredReplicas.
/// </summary>
public bool IsUnderReplicated => HasDesiredReplicas && Peers.Count < DesiredReplicas;
/// <summary>
/// True when the group has more peers than the desired replica count.
/// Go reference: jetstream_cluster.go extraPeers — len(rg.Peers) &gt; DesiredReplicas.
/// </summary>
public bool IsOverReplicated => HasDesiredReplicas && Peers.Count > DesiredReplicas;
/// <summary>
/// Returns true if the given peerId is a member of this group (case-sensitive).
/// Go reference: jetstream_cluster.go isMember helper.
/// </summary>
public bool IsMember(string peerId) => Peers.Contains(peerId, StringComparer.Ordinal);
/// <summary>
/// Sets the preferred leader peer for this group.
/// Throws <see cref="InvalidOperationException"/> if peerId is not a member.
/// Go reference: jetstream_cluster.go setPreferred / rg.Preferred assignment.
/// </summary>
public void SetPreferred(string peerId)
{
if (!IsMember(peerId))
throw new InvalidOperationException(
$"Cannot set preferred peer '{peerId}': not a member of RAFT group '{Name}'.");
Preferred = peerId;
}
/// <summary>
/// Removes a peer from the group. If the removed peer was the preferred peer,
/// <see cref="Preferred"/> is cleared. Returns true if the peer was found and removed.
/// Go reference: jetstream_cluster.go removePeer.
/// </summary>
public bool RemovePeer(string peerId)
{
var removed = Peers.Remove(peerId);
if (removed && string.Equals(Preferred, peerId, StringComparison.Ordinal))
Preferred = string.Empty;
return removed;
}
/// <summary>
/// Adds a peer to the group if not already present. Returns true if the peer was added.
/// Go reference: jetstream_cluster.go addPeer / expandGroup.
/// </summary>
public bool AddPeer(string peerId)
{
if (IsMember(peerId))
return false;
Peers.Add(peerId);
return true;
}
/// <summary>
/// Factory method that uses <see cref="PlacementEngine.SelectPeerGroup"/> to create a
/// <see cref="RaftGroup"/> with topology-aware peer selection and sets
/// <see cref="DesiredReplicas"/> to the requested replica count.
/// Go reference: jetstream_cluster.go createGroupForStream — calls selectPeerGroup then
/// assigns rg.DesiredReplicas = replicas.
/// </summary>
public static RaftGroup CreateRaftGroup(
string groupName,
int replicas,
IReadOnlyList<PeerInfo> availablePeers,
PlacementPolicy? policy = null)
{
var group = PlacementEngine.SelectPeerGroup(groupName, replicas, availablePeers, policy);
group.DesiredReplicas = replicas;
return group;
}
}
/// <summary>

260
tests/NATS.Server.Tests/JetStream/Cluster/RaftGroupLifecycleTests.cs Normal file
View File

@@ -0,0 +1,260 @@
// Go parity: golang/nats-server/server/jetstream_cluster.go
// Covers: RaftGroup member management, peer add/remove/preferred,
// factory method via PlacementEngine, replication health properties,
// and quorum size calculation.
using NATS.Server.JetStream.Cluster;
namespace NATS.Server.Tests.JetStream.Cluster;
/// <summary>
/// Tests for RaftGroup lifecycle: membership helpers, factory method,
/// replication status properties, and quorum size.
/// Go reference: jetstream_cluster.go:154-163 raftGroup struct and peer management.
/// </summary>
public class RaftGroupLifecycleTests
{
// ---------------------------------------------------------------
// IsMember — membership check
// Go reference: jetstream_cluster.go isMember helper
// ---------------------------------------------------------------
[Fact]
public void IsMember_returns_true_for_existing_peer()
{
// Go reference: jetstream_cluster.go isMember — checks rg.Peers contains id
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2", "peer-3"] };
group.IsMember("peer-2").ShouldBeTrue();
}
[Fact]
public void IsMember_returns_false_for_non_member()
{
// Go reference: jetstream_cluster.go isMember — returns false when not in Peers
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2"] };
group.IsMember("peer-9").ShouldBeFalse();
}
// ---------------------------------------------------------------
// SetPreferred — assign preferred peer
// Go reference: jetstream_cluster.go setPreferred / rg.Preferred
// ---------------------------------------------------------------
[Fact]
public void SetPreferred_sets_preferred_peer()
{
// Go reference: jetstream_cluster.go setPreferred — assigns rg.Preferred when member
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2", "peer-3"] };
group.SetPreferred("peer-3");
group.Preferred.ShouldBe("peer-3");
}
[Fact]
public void SetPreferred_throws_for_non_member()
{
// Go reference: jetstream_cluster.go setPreferred — validates membership before setting
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2"] };
Should.Throw<InvalidOperationException>(() => group.SetPreferred("peer-99"));
}
// ---------------------------------------------------------------
// RemovePeer — remove a peer from the group
// Go reference: jetstream_cluster.go removePeer
// ---------------------------------------------------------------
[Fact]
public void RemovePeer_removes_existing_peer()
{
// Go reference: jetstream_cluster.go removePeer — removes peer from rg.Peers
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2", "peer-3"] };
var removed = group.RemovePeer("peer-2");
removed.ShouldBeTrue();
group.Peers.ShouldNotContain("peer-2");
group.Peers.Count.ShouldBe(2);
}
[Fact]
public void RemovePeer_clears_preferred_when_removing_preferred()
{
// Go reference: jetstream_cluster.go removePeer — clears rg.Preferred if it matches removed peer
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2", "peer-3"], Preferred = "peer-2" };
group.RemovePeer("peer-2");
group.Preferred.ShouldBe(string.Empty);
}
[Fact]
public void RemovePeer_returns_false_for_non_member()
{
// Go reference: jetstream_cluster.go removePeer — returns false when peer not found
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2"] };
var removed = group.RemovePeer("peer-99");
removed.ShouldBeFalse();
group.Peers.Count.ShouldBe(2);
}
// ---------------------------------------------------------------
// AddPeer — add a new peer to the group
// Go reference: jetstream_cluster.go addPeer / expandGroup
// ---------------------------------------------------------------
[Fact]
public void AddPeer_adds_new_peer()
{
// Go reference: jetstream_cluster.go addPeer — appends peer to rg.Peers
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2"] };
var added = group.AddPeer("peer-3");
added.ShouldBeTrue();
group.Peers.ShouldContain("peer-3");
group.Peers.Count.ShouldBe(3);
}
[Fact]
public void AddPeer_returns_false_for_existing_peer()
{
// Go reference: jetstream_cluster.go addPeer — skips duplicate, returns false
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2"] };
var added = group.AddPeer("peer-1");
added.ShouldBeFalse();
group.Peers.Count.ShouldBe(2);
}
// ---------------------------------------------------------------
// CreateRaftGroup factory — uses PlacementEngine
// Go reference: jetstream_cluster.go:7212 selectPeerGroup called from createGroupForStream
// ---------------------------------------------------------------
[Fact]
public void CreateRaftGroup_uses_placement_engine()
{
// Go reference: jetstream_cluster.go createGroupForStream — calls selectPeerGroup
var peers = new List<PeerInfo>
{
new() { PeerId = "peer-A", Available = true, AvailableStorage = 9000 },
new() { PeerId = "peer-B", Available = true, AvailableStorage = 8000 },
new() { PeerId = "peer-C", Available = true, AvailableStorage = 7000 },
};
var group = RaftGroup.CreateRaftGroup("my-stream", 3, peers);
group.Name.ShouldBe("my-stream");
group.Peers.Count.ShouldBe(3);
group.Peers.ShouldContain("peer-A");
group.Peers.ShouldContain("peer-B");
group.Peers.ShouldContain("peer-C");
}
[Fact]
public void CreateRaftGroup_sets_desired_replicas()
{
// Go reference: jetstream_cluster.go rg.DesiredReplicas = replicas after group creation
var peers = new List<PeerInfo>
{
new() { PeerId = "peer-X", Available = true },
new() { PeerId = "peer-Y", Available = true },
new() { PeerId = "peer-Z", Available = true },
};
var group = RaftGroup.CreateRaftGroup("replicated-stream", 3, peers);
group.DesiredReplicas.ShouldBe(3);
group.HasDesiredReplicas.ShouldBeTrue();
}
// ---------------------------------------------------------------
// IsUnderReplicated — replication health
// Go reference: jetstream_cluster.go missingPeers — len(Peers) < DesiredReplicas
// ---------------------------------------------------------------
[Fact]
public void IsUnderReplicated_true_when_peers_less_than_desired()
{
// Go reference: jetstream_cluster.go:2284 sa.missingPeers()
var group = new RaftGroup { Name = "test", Peers = ["peer-1"], DesiredReplicas = 3 };
group.IsUnderReplicated.ShouldBeTrue();
}
[Fact]
public void IsUnderReplicated_false_when_peers_equal_desired()
{
// Go reference: jetstream_cluster.go:2284 sa.missingPeers() — no deficit when equal
var group = new RaftGroup { Name = "test", Peers = ["peer-1", "peer-2", "peer-3"], DesiredReplicas = 3 };
group.IsUnderReplicated.ShouldBeFalse();
}
[Fact]
public void IsUnderReplicated_false_when_no_desired_replicas_set()
{
// Go reference: jetstream_cluster.go — without DesiredReplicas set, no under-replication
var group = new RaftGroup { Name = "test", Peers = ["peer-1"] };
group.IsUnderReplicated.ShouldBeFalse();
}
// ---------------------------------------------------------------
// IsOverReplicated — excess replication detection
// Go reference: jetstream_cluster.go extraPeers — len(Peers) > DesiredReplicas
// ---------------------------------------------------------------
[Fact]
public void IsOverReplicated_true_when_peers_more_than_desired()
{
// Go reference: jetstream_cluster.go extraPeers detection for scale-down
var group = new RaftGroup { Name = "test", Peers = ["p1", "p2", "p3", "p4"], DesiredReplicas = 3 };
group.IsOverReplicated.ShouldBeTrue();
}
[Fact]
public void IsOverReplicated_false_when_peers_equal_desired()
{
// Go reference: jetstream_cluster.go — no excess when equal
var group = new RaftGroup { Name = "test", Peers = ["p1", "p2", "p3"], DesiredReplicas = 3 };
group.IsOverReplicated.ShouldBeFalse();
}
[Fact]
public void IsOverReplicated_false_when_no_desired_replicas_set()
{
// Go reference: jetstream_cluster.go — without DesiredReplicas set, no over-replication
var group = new RaftGroup { Name = "test", Peers = ["p1", "p2", "p3", "p4"] };
group.IsOverReplicated.ShouldBeFalse();
}
// ---------------------------------------------------------------
// QuorumSize — majority quorum calculation
// Go reference: jetstream_cluster.go quorumNeeded — (n/2)+1
// ---------------------------------------------------------------
[Theory]
[InlineData(1, 1)] // R=1 → quorum=1
[InlineData(3, 2)] // R=3 → quorum=2
[InlineData(5, 3)] // R=5 → quorum=3
[InlineData(2, 2)] // R=2 → quorum=2 (degenerate, but formula consistent)
[InlineData(4, 3)] // R=4 → quorum=3
public void QuorumSize_correct_for_various_counts(int peerCount, int expectedQuorum)
{
// Go reference: jetstream_cluster.go quorumNeeded — (n/2)+1
var peers = Enumerable.Range(1, peerCount).Select(i => $"peer-{i}").ToList();
var group = new RaftGroup { Name = "quorum-test", Peers = peers };
group.QuorumSize.ShouldBe(expectedQuorum);
}
}