Commit phase - AKKA Model

neo/Consensus/CommitAgreement.cs

@@ -0,0 +1,30 @@
+using System.IO;
+using Neo.IO;
+
+namespace Neo.Consensus
+{
+    internal class CommitAgreement : ConsensusMessage
+    {
+        /// <summary>
+        /// Block hash of the signature
+        /// </summary>
+        public UInt256 BlockHash;
+
+        /// <summary>
+        /// Constructors
+        /// </summary>
+        public CommitAgreement() : base(ConsensusMessageType.CommitAgreement) { }
+
+        public override void Deserialize(BinaryReader reader)
+        {
+            base.Deserialize(reader);
+            BlockHash = reader.ReadSerializable<UInt256>();
+        }
+
+        public override void Serialize(BinaryWriter writer)
+        {
+            base.Serialize(writer);
+            writer.Write(BlockHash);
+        }
+    }
+}

neo/Consensus/ConsensusContext.cs

@@ -31,20 +31,42 @@ internal class ConsensusContext : IDisposable
         public byte[] ExpectedView;
         public KeyPair KeyPair;
 
+        private UInt256[] Commits;
+        private Block _header = null;
+        public UInt256 CommitHash => _header?.Hash;
+
         public int M => Validators.Length - (Validators.Length - 1) / 3;
 
+        public bool TryToCommit(ConsensusPayload payload, CommitAgreement message)
+        {
+            // Already received
+            if (Commits[payload.ValidatorIndex] != null) return false;
+
+            // Store received block hash
+            Commits[payload.ValidatorIndex] = message.BlockHash;
+
+            // Check count
+            return _header != null && Commits.Where(u => u != null && u == _header.Hash).Count() >= M;
+        }
+
         public void ChangeView(byte view_number)
         {
             State &= ConsensusState.SignatureSent;
             ViewNumber = view_number;
             PrimaryIndex = GetPrimaryIndex(view_number);
+
             if (State == ConsensusState.Initial)
             {
                 TransactionHashes = null;
                 Signatures = new byte[Validators.Length][];
+                Commits = new UInt256[Validators.Length];
             }
+
             if (MyIndex >= 0)
+            {
                 ExpectedView[MyIndex] = view_number;
+            }
+
             _header = null;
         }
 
@@ -67,10 +89,10 @@ public ConsensusPayload MakeChangeView()
             });
         }
 
-        private Block _header = null;
         public Block MakeHeader()
         {
             if (TransactionHashes == null) return null;
+
             if (_header == null)
             {
                 _header = new Block
@@ -84,10 +106,23 @@ public Block MakeHeader()
                     NextConsensus = NextConsensus,
                     Transactions = new Transaction[0]
                 };
+
+                Commits[MyIndex] = _header.Hash;
             }
+
             return _header;
         }
 
+        public ConsensusPayload MakeCommitAgreement()
+        {
+            if (_header == null) return null;
+
+            return MakePayload(new CommitAgreement()
+            {
+                BlockHash = _header.Hash
+            });
+        }
+
         private ConsensusPayload MakePayload(ConsensusMessage message)
         {
             message.ViewNumber = ViewNumber;
@@ -137,6 +172,8 @@ public void Reset(Wallet wallet)
             Signatures = new byte[Validators.Length][];
             ExpectedView = new byte[Validators.Length];
             KeyPair = null;
+            Commits = new UInt256[Validators.Length];
+
             for (int i = 0; i < Validators.Length; i++)
             {
                 WalletAccount account = wallet.GetAccount(Validators[i]);

neo/Consensus/ConsensusMessageType.cs

@@ -10,5 +10,7 @@ internal enum ConsensusMessageType : byte
         PrepareRequest = 0x20,
         [ReflectionCache(typeof(PrepareResponse))]
         PrepareResponse = 0x21,
+        [ReflectionCache(typeof(CommitAgreement))]
+        CommitAgreement = 0x22,
     }
 }

neo/Consensus/ConsensusService.cs

@@ -74,6 +74,18 @@ private void ChangeTimer(TimeSpan delay)
         private void CheckExpectedView(byte view_number)
         {
             if (context.ViewNumber == view_number) return;
+            if (context.State.HasFlag(ConsensusState.CommitSent))
+            {
+                if (context.State.HasFlag(ConsensusState.SignatureSent))
+                {
+                    // If signature was sent, we send again
+
+                    SignAndRelay(context.MakePrepareResponse(context.Signatures[context.MyIndex]));
+                }
+
+                return;
+            }
+
             if (context.ExpectedView.Count(p => p == view_number) >= context.M)
             {
                 InitializeConsensus(view_number);
@@ -82,8 +94,30 @@ private void CheckExpectedView(byte view_number)
 
         private void CheckSignatures()
         {
+            if (!context.State.HasFlag(ConsensusState.CommitSent) &&
+                context.Signatures.Count(p => p != null) >= context.M &&
+                context.TransactionHashes.All(p => context.Transactions.ContainsKey(p)))
+            {
+                // Send my commit
+
+                context.State |= ConsensusState.CommitSent;
+                SignAndRelay(context.MakeCommitAgreement());
+
+                Log($"Commit sent: height={context.BlockIndex} hash={context.CommitHash} state={context.State}");
+            }
+        }
+
+        private void OnCommitAgreement(ConsensusPayload payload, CommitAgreement message)
+        {
+            if (context.State.HasFlag(ConsensusState.BlockSent) ||
+                !context.TryToCommit(payload, message)) return;
+
+            Log($"{nameof(OnCommitAgreement)}: height={payload.BlockIndex} hash={message.BlockHash.ToString()} view={message.ViewNumber} index={payload.ValidatorIndex}");
+
             if (context.Signatures.Count(p => p != null) >= context.M && context.TransactionHashes.All(p => context.Transactions.ContainsKey(p)))
             {
+                context.State |= ConsensusState.BlockSent;
+
                 Contract contract = Contract.CreateMultiSigContract(context.M, context.Validators);
                 Block block = context.MakeHeader();
                 ContractParametersContext sc = new ContractParametersContext(block);
@@ -96,8 +130,8 @@ private void CheckSignatures()
                 sc.Verifiable.Witnesses = sc.GetWitnesses();
                 block.Transactions = context.TransactionHashes.Select(p => context.Transactions[p]).ToArray();
                 Log($"relay block: {block.Hash}");
+
                 system.LocalNode.Tell(new LocalNode.Relay { Inventory = block });
-                context.State |= ConsensusState.BlockSent;
             }
         }
 
@@ -178,18 +212,21 @@ private void Log(string message, LogLevel level = LogLevel.Info)
 
         private void OnChangeViewReceived(ConsensusPayload payload, ChangeView message)
         {
-            Log($"{nameof(OnChangeViewReceived)}: height={payload.BlockIndex} view={message.ViewNumber} index={payload.ValidatorIndex} nv={message.NewViewNumber}");
             if (message.NewViewNumber <= context.ExpectedView[payload.ValidatorIndex])
                 return;
+
+            Log($"{nameof(OnChangeViewReceived)}: height={payload.BlockIndex} view={message.ViewNumber} index={payload.ValidatorIndex} nv={message.NewViewNumber}");
+
             context.ExpectedView[payload.ValidatorIndex] = message.NewViewNumber;
             CheckExpectedView(message.NewViewNumber);
         }
 
         private void OnConsensusPayload(ConsensusPayload payload)
         {
-            if (payload.ValidatorIndex == context.MyIndex) return;
-            if (payload.Version != ConsensusContext.Version)
-                return;
+            if (payload.ValidatorIndex == context.MyIndex ||
+                payload.Version != ConsensusContext.Version ||
+                payload.ValidatorIndex >= context.Validators.Length) return;
+
             if (payload.PrevHash != context.PrevHash || payload.BlockIndex != context.BlockIndex)
             {
                 if (context.Snapshot.Height + 1 < payload.BlockIndex)
@@ -198,7 +235,7 @@ private void OnConsensusPayload(ConsensusPayload payload)
                 }
                 return;
             }
-            if (payload.ValidatorIndex >= context.Validators.Length) return;
+
             ConsensusMessage message;
             try
             {
@@ -208,8 +245,10 @@ private void OnConsensusPayload(ConsensusPayload payload)
             {
                 return;
             }
+
             if (message.ViewNumber != context.ViewNumber && message.Type != ConsensusMessageType.ChangeView)
                 return;
+
             switch (message.Type)
             {
                 case ConsensusMessageType.ChangeView:
@@ -221,6 +260,9 @@ private void OnConsensusPayload(ConsensusPayload payload)
                 case ConsensusMessageType.PrepareResponse:
                     OnPrepareResponseReceived(payload, (PrepareResponse)message);
                     break;
+                case ConsensusMessageType.CommitAgreement:
+                    OnCommitAgreement(payload, (CommitAgreement)message);
+                    break;
             }
         }
 
@@ -233,15 +275,22 @@ private void OnPersistCompleted(Block block)
 
         private void OnPrepareRequestReceived(ConsensusPayload payload, PrepareRequest message)
         {
+            if (context.State.HasFlag(ConsensusState.RequestReceived))
+                return;
+
             Log($"{nameof(OnPrepareRequestReceived)}: height={payload.BlockIndex} view={message.ViewNumber} index={payload.ValidatorIndex} tx={message.TransactionHashes.Length}");
-            if (!context.State.HasFlag(ConsensusState.Backup) || context.State.HasFlag(ConsensusState.RequestReceived))
+
+            if (!context.State.HasFlag(ConsensusState.Backup))
                 return;
+
             if (payload.ValidatorIndex != context.PrimaryIndex) return;
+
             if (payload.Timestamp <= context.Snapshot.GetHeader(context.PrevHash).Timestamp || payload.Timestamp > DateTime.UtcNow.AddMinutes(10).ToTimestamp())
             {
                 Log($"Timestamp incorrect: {payload.Timestamp}", LogLevel.Warning);
                 return;
             }
+
             context.State |= ConsensusState.RequestReceived;
             context.Timestamp = payload.Timestamp;
             context.Nonce = message.Nonce;
@@ -271,9 +320,11 @@ private void OnPrepareRequestReceived(ConsensusPayload payload, PrepareRequest m
 
         private void OnPrepareResponseReceived(ConsensusPayload payload, PrepareResponse message)
         {
-            Log($"{nameof(OnPrepareResponseReceived)}: height={payload.BlockIndex} view={message.ViewNumber} index={payload.ValidatorIndex}");
             if (context.State.HasFlag(ConsensusState.BlockSent)) return;
             if (context.Signatures[payload.ValidatorIndex] != null) return;
+
+            Log($"{nameof(OnPrepareResponseReceived)}: height={payload.BlockIndex} view={message.ViewNumber} index={payload.ValidatorIndex}");
+
             Block header = context.MakeHeader();
             if (header == null || !Crypto.Default.VerifySignature(header.GetHashData(), message.Signature, context.Validators[payload.ValidatorIndex].EncodePoint(false))) return;
             context.Signatures[payload.ValidatorIndex] = message.Signature;
@@ -360,6 +411,20 @@ public static Props Props(NeoSystem system, Wallet wallet)
 
         private void RequestChangeView()
         {
+            if (context.State.HasFlag(ConsensusState.CommitSent))
+            {
+                // Lock view change on timer
+
+                if (context.State.HasFlag(ConsensusState.SignatureSent))
+                {
+                    // If signature was sent, we send again
+
+                    SignAndRelay(context.MakePrepareResponse(context.Signatures[context.MyIndex]));
+                }
+
+                return;
+            }
+
             context.State |= ConsensusState.ViewChanging;
             context.ExpectedView[context.MyIndex]++;
             Log($"request change view: height={context.BlockIndex} view={context.ViewNumber} nv={context.ExpectedView[context.MyIndex]} state={context.State}");
@@ -370,6 +435,8 @@ private void RequestChangeView()
 
         private void SignAndRelay(ConsensusPayload payload)
         {
+            if (payload == null) return;
+
             ContractParametersContext sc;
             try
             {
@@ -380,6 +447,7 @@ private void SignAndRelay(ConsensusPayload payload)
             {
                 return;
             }
+
             sc.Verifiable.Witnesses = sc.GetWitnesses();
             system.LocalNode.Tell(new LocalNode.SendDirectly { Inventory = payload });
         }

neo/Consensus/ConsensusState.cs

@@ -13,5 +13,6 @@ internal enum ConsensusState : byte
         SignatureSent = 0x10,
         BlockSent = 0x20,
         ViewChanging = 0x40,
+        CommitSent = 0x80,
     }
-}
+}