Ticket #102 : Fix consensus protocol

habarthierry-hue · habarthierry-hue · commit 8657936530fd · 2022-05-10T21:15:36.000+02:00
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/BaseNodeHost.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/BaseNodeHost.cs
@@ -123,7 +123,7 @@ protected async Task HandleUDPPackage()
             try
             {
                 var udpResult = await UdpServer.ReceiveAsync().WithCancellation(TokenSource.Token);
-                // if (await TryHandleConsensusRequest(udpResult)) return;
+                if (await TryHandleConsensusRequest(udpResult)) return;
                 if (await TryHandleGossipRequest(udpResult)) return;
                 await HandleUDPPackage(udpResult, TokenSource.Token);
             }
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/BasePeerHost.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/BasePeerHost.cs
@@ -37,7 +37,7 @@ public abstract class BasePeerHost : IPeerHost
     {
         private readonly ILogger<BasePeerHost> _logger;
         private readonly ConsensusPeerOptions _options;
-        private readonly ClusterStore _clusterStore;
+        private readonly IClusterStore _clusterStore;
         private readonly ILogStore _logStore;
         private readonly IPeerStore _peerStore;
         private readonly string _peerId;
@@ -51,7 +51,7 @@ public abstract class BasePeerHost : IPeerHost
         private System.Timers.Timer _electionCheckTimer;
         private System.Timers.Timer _leaderHeartbeatTimer;
 
-        public BasePeerHost(ILogger<BasePeerHost> logger, IOptions<ConsensusPeerOptions> options, ClusterStore clusterStore, ILogStore logStore, IPeerStore peerStore)
+        public BasePeerHost(ILogger<BasePeerHost> logger, IOptions<ConsensusPeerOptions> options, IClusterStore clusterStore, ILogStore logStore, IPeerStore peerStore)
         {
             _logger = logger;
             _options = options.Value;
@@ -157,6 +157,7 @@ private async Task StartElection()
             // Start to vote.
             _logger.LogInformation("{Node}:{PeerId}:{TermId}, Start to vote", _nodeId, _peerId, Info.TermId);
             var nodes = await _clusterStore.GetAllNodes(TokenSource.Token);
+            nodes = nodes.Where(n => n.Port != _options.Port || n.Url != _options.Url);
             _quorum = (nodes.Count() / 2) + 1;
             if (_quorum == 0)
             {
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Core/StandalonePeerHost.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Core/StandalonePeerHost.cs
@@ -9,7 +9,7 @@ namespace FaasNet.RaftConsensus.Core
 {
     public class StandalonePeerHost : BasePeerHost
     {
-        public StandalonePeerHost(ILogger<BasePeerHost> logger, IOptions<ConsensusPeerOptions> options, ClusterStore clusterStore, ILogStore logStore, IPeerStore peerStore) : base(logger, options, clusterStore, logStore, peerStore)
+        public StandalonePeerHost(ILogger<BasePeerHost> logger, IOptions<ConsensusPeerOptions> options, IClusterStore clusterStore, ILogStore logStore, IPeerStore peerStore) : base(logger, options, clusterStore, logStore, peerStore)
         {
         }
 
diff --git a/src/RaftConsensus/FaasNet.RaftConsensus.Logging.Startup/Program.cs b/src/RaftConsensus/FaasNet.RaftConsensus.Logging.Startup/Program.cs
@@ -3,65 +3,44 @@
 using FaasNet.RaftConsensus.Core;
 using FaasNet.RaftConsensus.Core.Models;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using System.Collections.Concurrent;
 
 int seedPort = 4000;
-var allNodes = CreateNodes();
+var peerInfos = AddPeers();
+var allNodes = CreateNodes(peerInfos);
 await StartNodes(allNodes);
 await DisplayMenu(allNodes);
+Console.WriteLine("Press Enter to quit the application");
+Console.ReadLine();
 
-/*
-var peerInfos = new ConcurrentBag<string>();
-do
-{
-    Console.WriteLine("Enter 'Q' to stop adding new peer or enter the name of your peer");
-    string str = Console.ReadLine();
-    continueExecution = str != "Q";
-    if (continueExecution) peerInfos.Add(str);
-}
-while (continueExecution);
+foreach (var node in allNodes) await node.Stop();
 
-continueExecution = true;
-do
+ConcurrentBag<string> AddPeers()
 {
-    Console.WriteLine("Enter 'Q' to stop sending message / Enter 'inf' to display the status of the peers / Enter the name of the partition");
-    string termId = Console.ReadLine();
-    continueExecution = termId != "Q";
-    if(termId == "inf")
+    var continueExecution = true;
+    var peerInfos = new ConcurrentBag<string>();
+    do
     {
-        foreach(var node in allNodes)
-        {
-            foreach(var peer in node.Peers)
-            {
-                Console.WriteLine($"Node '{node.NodeId}', Peer '{peer.PeerId}', TermId '{peer.Info.TermId}', TermIndex '{peer.Info.ConfirmedTermIndex}', State '{peer.State}");
-            }
-        }
-
-        continue;
+        Console.WriteLine("Enter 'Q' to stop adding new peer");
+        Console.WriteLine("Enter the name of your peer");
+        string str = Console.ReadLine();
+        continueExecution = str != "Q";
+        if (continueExecution) peerInfos.Add(str);
     }
-
-    Console.WriteLine("Please enter the message");
-    string message = Console.ReadLine();
-    var consensusClient = new ConsensusClient("localhost", seedPort);
-    await consensusClient.AppendEntry(termId, message, CancellationToken.None);
+    while (continueExecution);
+    return peerInfos;
 }
-while (continueExecution);
-*/
-
-Console.WriteLine("Press Enter to quit the application");
-Console.ReadLine();
-
-foreach (var node in allNodes) await node.Stop();
 
-ICollection<INodeHost> CreateNodes()
+ICollection<INodeHost> CreateNodes(ConcurrentBag<string> peerInfos)
 {
     Console.WriteLine("How many nodes do you want to start ?");
     int nbNodes = int.Parse(Console.ReadLine());
     var allNodes = new List<INodeHost>();
     for (int i = 0; i <= nbNodes; i++)
     {
         allNodes.Add(BuildNodeHost(
-            new ConcurrentBag<PeerInfo>(/*peerInfos.Select(p => new PeerInfo { TermId = p })*/),
+            new ConcurrentBag<PeerInfo>(peerInfos.Select(p => new PeerInfo { TermId = p })),
             seedPort + i,
             Path.Combine(Directory.GetCurrentDirectory(), i + "-log-{0}.txt"),
             i == 0));
@@ -104,10 +83,15 @@ async Task DisplayMenu(ICollection<INodeHost> nodes)
     var continueExecution = true;
     do
     {
-        Console.WriteLine("Enter 'Q' to stop execution / Enter 'inf' to display the states / Start a new node 'addnode' / Stop one node 'stopnode'");
-        string termId = Console.ReadLine();
-        continueExecution = termId != "Q";
-        if (termId == "inf")
+        Console.WriteLine("Enter 'Q' to stop execution");
+        Console.WriteLine("Enter 'states' to display the states");
+        Console.WriteLine("Enter 'peers' to display the status of the peers");
+        Console.WriteLine("Enter 'appendentry' to append an entry");
+        Console.WriteLine("Start a new node 'addnode'");
+        Console.WriteLine("Stop one node 'stopnode'");
+        string menuId = Console.ReadLine();
+        continueExecution = menuId != "Q";
+        if (menuId == "states")
         {
             foreach (var node in nodes)
             {
@@ -121,10 +105,10 @@ async Task DisplayMenu(ICollection<INodeHost> nodes)
             continue;
         }
 
-        if(termId == "addnode")
+        if(menuId == "addnode")
         {
             var newNode = BuildNodeHost(
-                new ConcurrentBag<PeerInfo>(/*peerInfos.Select(p => new PeerInfo { TermId = p })*/),
+                new ConcurrentBag<PeerInfo>(peerInfos.Select(p => new PeerInfo { TermId = p })),
                 seedPort + nodes.Count(),
                 Path.Combine(Directory.GetCurrentDirectory(), nodes.Count() + "-log-{0}.txt"),
                 false);
@@ -133,13 +117,35 @@ async Task DisplayMenu(ICollection<INodeHost> nodes)
             continue;
         }
 
-        if (termId == "stopnode")
+        if (menuId == "stopnode")
         {
             Console.WriteLine("Enter the index");
             int index = int.Parse(Console.ReadLine());
             await nodes.ElementAt(index).Stop();
             continue;
         }
+
+        if (menuId == "peers")
+        {
+            foreach (var node in allNodes)
+            {
+                foreach (var peer in node.Peers)
+                {
+                    Console.WriteLine($"Node '{node.NodeId}', Peer '{peer.PeerId}', TermId '{peer.Info.TermId}', TermIndex '{peer.Info.ConfirmedTermIndex}', State '{peer.State}");
+                }
+            }
+            continue;
+        }
+
+        if(menuId == "appendentry")
+        {
+            Console.WriteLine("Enter the termid");
+            string termId = Console.ReadLine();
+            Console.WriteLine("Please enter the message");
+            string message = Console.ReadLine();
+            var consensusClient = new ConsensusClient("localhost", seedPort);
+            await consensusClient.AppendEntry(termId, message, CancellationToken.None);
+        }
     }
     while (continueExecution);
 }
@@ -149,9 +155,8 @@ INodeHost BuildNodeHost(ConcurrentBag<PeerInfo> peers, int port, string logPath,
     var services = new ServiceCollection()
         .AddConsensusPeer(o => o.Port = port);
     if (isSeed) services.SetNodeStates(new ConcurrentBag<NodeState> { new ClusterNode { Port = 4000, Url = "localhost" }.ToNodeState() });
-        // .SetPeers(peers)
-        // .SetNodeStates(nodeStates)
     var serviceProvider = services.UseLogFileStore(o => o.LogFilePath = logPath)
+        .SetPeers(peers)
         .Services
         // .AddLogging(l => l.AddConsole())
         .BuildServiceProvider();
diff --git a/tests/FaasNet.RaftConsensus.Tests/PeerHostFixture.cs b/tests/FaasNet.RaftConsensus.Tests/PeerHostFixture.cs
@@ -49,13 +49,13 @@ public async Task When_JoinNodeToCluster_Then_StorageIsUpdated()
         public async Task When_NodeIsStopped_Then_NodeBecomeUnreachable()
         {
             // ARRANGE
-            var seedNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4000, new ConcurrentBag<ClusterNode>(), true);
-            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4001, new ConcurrentBag<ClusterNode>());
+            var seedNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4001, new ConcurrentBag<ClusterNode>(), true);
+            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4002, new ConcurrentBag<ClusterNode>());
             await seedNode.Start(CancellationToken.None);
             await firstNode.Start(CancellationToken.None);
             WaitNodeIsStarted(seedNode);
             WaitNodeIsStarted(firstNode);
-            using (var gossipClient = new GossipClient("localhost", 4000)) await gossipClient.JoinNode("localhost", 4001);
+            using (var gossipClient = new GossipClient("localhost", 4001)) await gossipClient.JoinNode("localhost", 4002);
             var seedNodeStates = await WaitEntityTypes(seedNode, (nodes) => nodes.Count() == 2);
             var firstNodeStates = await WaitEntityTypes(firstNode, (nodes) => nodes.Count() == 2);
 
@@ -67,25 +67,25 @@ public async Task When_NodeIsStopped_Then_NodeBecomeUnreachable()
             // ASSERT
             Assert.Single(clusterNodes);
             Assert.Equal("localhost", clusterNodes.First().Node.Url);
-            Assert.Equal(4001, clusterNodes.First().Node.Port);
+            Assert.Equal(4002, clusterNodes.First().Node.Port);
         }
 
         [Fact]
         public async Task When_StateIsAdded_Then_StorageIsUpdated()
         {
             // ARRANGE
-            var seedNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4000, new ConcurrentBag<ClusterNode>(), true);
-            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4001, new ConcurrentBag<ClusterNode>());
+            var seedNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4003, new ConcurrentBag<ClusterNode>(), true);
+            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4004, new ConcurrentBag<ClusterNode>());
             await seedNode.Start(CancellationToken.None);
             await firstNode.Start(CancellationToken.None);
             WaitNodeIsStarted(seedNode);
             WaitNodeIsStarted(firstNode);
-            using (var gossipClient = new GossipClient("localhost", 4000)) await gossipClient.JoinNode("localhost", 4001);
+            using (var gossipClient = new GossipClient("localhost", 4003)) await gossipClient.JoinNode("localhost", 4004);
             var seedNodeStates = await WaitEntityTypes(seedNode, (nodes) => nodes.Count() == 2);
             var firstNodeStates = await WaitEntityTypes(firstNode, (nodes) => nodes.Count() == 2);
 
             // ACT
-            using (var gossipClient = new GossipClient("localhost", 4000)) await gossipClient.UpdateNodeState("Client", "id", "value");
+            using (var gossipClient = new GossipClient("localhost", 4003)) await gossipClient.UpdateNodeState("Client", "id", "value");
             var seedClient = (await WaitEntityTypes(seedNode, (nodes) => nodes.Any(n => n.EntityType == "Client"))).First(c => c.EntityType == "Client");
             var firstNodeClient = (await WaitEntityTypes(firstNode, (nodes) => nodes.Any(n => n.EntityType == "Client"))).First(c => c.EntityType == "Client");
 
@@ -98,105 +98,77 @@ public async Task When_StateIsAdded_Then_StorageIsUpdated()
 
         #endregion
 
+        #region Consensus
+
         [Fact]
         public async Task When_AppendLogInOnePartition_Then_LogIsReplicated()
         {
             // ARRANGE
-            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo>
-            {
-                new PeerInfo { TermId = "termId", TermIndex = 0 }
-            }, 4001, new ConcurrentBag<ClusterNode>
-            {
-                new ClusterNode
-                {
-                    Port = 4002,
-                    Url = "localhost"
-                }
-            });
-            var secondNode = BuildNodeHost(new ConcurrentBag<PeerInfo>
-            {
-                new PeerInfo { TermId = "termId", TermIndex = 0 }
-            }, 4002, new ConcurrentBag<ClusterNode>
-            {
-                new ClusterNode
-                {
-                    Port = 4001,
-                    Url = "localhost"
-                }
-            });
+            var seedNode = BuildNodeHost(new ConcurrentBag<PeerInfo> { new PeerInfo { TermId = "termId", TermIndex = 0 } }, 4005, new ConcurrentBag<ClusterNode>(), true);
+            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo> { new PeerInfo { TermId = "termId", TermIndex = 0 } }, 4006, new ConcurrentBag<ClusterNode>());
+            await seedNode.Start(CancellationToken.None);
             await firstNode.Start(CancellationToken.None);
-            await secondNode.Start(CancellationToken.None);
-            var allNodes = new List<INodeHost> { firstNode, secondNode };
+            WaitNodeIsStarted(seedNode);
+            WaitNodeIsStarted(firstNode);
+            using (var gossipClient = new GossipClient("localhost", 4005)) await gossipClient.JoinNode("localhost", 4006);
+            var seedNodeStates = await WaitEntityTypes(seedNode, (nodes) => nodes.Count() == 2);
+            var firstNodeStates = await WaitEntityTypes(firstNode, (nodes) => nodes.Count() == 2);
+            var allNodes = new List<INodeHost> { seedNode, firstNode };
 
             // ACT
             WaitOnlyOneLeader(allNodes, "termId");
-            var client = new ConsensusClient("localhost", 4001);
+            var client = new ConsensusClient("localhost", 4005);
             client.AppendEntry("termId", "value", CancellationToken.None).Wait();
             WaitLogs(allNodes, p => p.Info.TermId == "termId", l => l.Value == "value");
 
             // ASSERT
+            var seedPeerLogs = seedNode.Peers.First().LogStore.GetAll(CancellationToken.None).Result;
             var firstPeerLogs = firstNode.Peers.First().LogStore.GetAll(CancellationToken.None).Result;
-            var secondPeerLogs = secondNode.Peers.First().LogStore.GetAll(CancellationToken.None).Result;
+            Assert.Single(seedPeerLogs);
             Assert.Single(firstPeerLogs);
-            Assert.Single(secondPeerLogs);
+            Assert.Equal("value", seedPeerLogs.First().Value);
             Assert.Equal("value", firstPeerLogs.First().Value);
-            Assert.Equal("value", secondPeerLogs.First().Value);
+            await seedNode.Stop();
             await firstNode.Stop();
-            await secondNode.Stop();
         }
 
         [Fact]
         public async Task When_AppendLogInTwoPartitions_Then_LogIsReplicated()
         {
             // ARRANGE
-            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo>
-            {
-                new PeerInfo { TermId = "termId", TermIndex = 0 },
-                new PeerInfo { TermId = "secondTermId", TermIndex = 0 }
-            }, 4001, new ConcurrentBag<ClusterNode>
-            {
-                new ClusterNode
-                {
-                    Port = 4002,
-                    Url = "localhost"
-                }
-            });
-            var secondNode = BuildNodeHost(new ConcurrentBag<PeerInfo>
-            {
-                new PeerInfo { TermId = "termId", TermIndex = 0 },
-                new PeerInfo { TermId = "secondTermId", TermIndex = 0 }
-            }, 4002, new ConcurrentBag<ClusterNode>
-            {
-                new ClusterNode
-                {
-                    Port = 4001,
-                    Url = "localhost"
-                }
-            });
+            var seedNode = BuildNodeHost(new ConcurrentBag<PeerInfo> { new PeerInfo { TermId = "termId", TermIndex = 0 }, new PeerInfo { TermId = "secondTermId", TermIndex = 0 } }, 4007, new ConcurrentBag<ClusterNode>(), true);
+            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo> { new PeerInfo { TermId = "termId", TermIndex = 0 }, new PeerInfo { TermId = "secondTermId", TermIndex = 0 } }, 4008, new ConcurrentBag<ClusterNode>());
+            await seedNode.Start(CancellationToken.None);
             await firstNode.Start(CancellationToken.None);
-            await secondNode.Start(CancellationToken.None);
-            var allNodes = new List<INodeHost> { firstNode, secondNode };
+            WaitNodeIsStarted(seedNode);
+            WaitNodeIsStarted(firstNode);
+            using (var gossipClient = new GossipClient("localhost", 4007)) await gossipClient.JoinNode("localhost", 4008);
+            var seedNodeStates = await WaitEntityTypes(seedNode, (nodes) => nodes.Count() == 2);
+            var firstNodeStates = await WaitEntityTypes(firstNode, (nodes) => nodes.Count() == 2);
+            var allNodes = new List<INodeHost> { seedNode, firstNode };
 
             // ACT
             WaitOnlyOneLeader(allNodes, "termId");
             WaitOnlyOneLeader(allNodes, "secondTermId");
-            var client = new ConsensusClient("localhost", 4001);
+            var client = new ConsensusClient("localhost", 4007);
             client.AppendEntry("termId", "value", CancellationToken.None).Wait();
             client.AppendEntry("secondTermId", "value", CancellationToken.None).Wait();
             WaitLogs(allNodes, p => p.Info.TermId == "termId", l => l.Value == "value");
             WaitLogs(allNodes, p => p.Info.TermId == "secondTermId", l => l.Value == "value");
 
             // ASSERT
+            var seedPeerLogs = seedNode.Peers.First().LogStore.GetAll(CancellationToken.None).Result;
             var firstPeerLogs = firstNode.Peers.First().LogStore.GetAll(CancellationToken.None).Result;
-            var secondPeerLogs = secondNode.Peers.First().LogStore.GetAll(CancellationToken.None).Result;
+            Assert.Single(seedPeerLogs);
             Assert.Single(firstPeerLogs);
-            Assert.Single(secondPeerLogs);
+            Assert.Equal("value", seedPeerLogs.First().Value);
             Assert.Equal("value", firstPeerLogs.First().Value);
-            Assert.Equal("value", secondPeerLogs.First().Value);
+            await seedNode.Stop();
             await firstNode.Stop();
-            await secondNode.Stop();
         }
 
+        #endregion
+
         private static INodeHost BuildNodeHost(ConcurrentBag<PeerInfo> peers, int port, ConcurrentBag<ClusterNode> clusterNodes, bool isSeed = false)
         {
             var serviceCollection = new ServiceCollection()

Original file line number	Diff line number	Diff line change
`@@ -123,7 +123,7 @@ protected async Task HandleUDPPackage()`
`123`	`123`	`try`
`124`	`124`	`{`
`125`	`125`	`var udpResult = await UdpServer.ReceiveAsync().WithCancellation(TokenSource.Token);`
`126`		`- // if (await TryHandleConsensusRequest(udpResult)) return;`
	`126`	`+ if (await TryHandleConsensusRequest(udpResult)) return;`
`127`	`127`	`if (await TryHandleGossipRequest(udpResult)) return;`
`128`	`128`	`await HandleUDPPackage(udpResult, TokenSource.Token);`
`129`	`129`	`}`
Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,7 @@ namespace FaasNet.RaftConsensus.Core`
`9`	`9`	`{`
`10`	`10`	`public class StandalonePeerHost : BasePeerHost`
`11`	`11`	`{`
`12`		`- public StandalonePeerHost(ILogger<BasePeerHost> logger, IOptions<ConsensusPeerOptions> options, ClusterStore clusterStore, ILogStore logStore, IPeerStore peerStore) : base(logger, options, clusterStore, logStore, peerStore)`
	`12`	`+ public StandalonePeerHost(ILogger<BasePeerHost> logger, IOptions<ConsensusPeerOptions> options, IClusterStore clusterStore, ILogStore logStore, IPeerStore peerStore) : base(logger, options, clusterStore, logStore, peerStore)`
`13`	`13`	`{`
`14`	`14`	`}`
`15`	`15`