Cache up-to-date consensus payloads

neo.UnitTests/Consensus/UT_Consensus.cs

@@ -209,6 +209,11 @@ public void TestSerializeAndDeserializeConsensusContext()
 
             consensusContext.LastChangeViewPayloads = new ConsensusPayload[consensusContext.Validators.Length];
 
+            consensusContext.FutureCommitPayloads = new ConsensusPayload[consensusContext.Validators.Length];
+            consensusContext.FuturePreparationPayloads = new ConsensusPayload[consensusContext.Validators.Length];
+            consensusContext.FutureChangeViewPayloads = new ConsensusPayload[consensusContext.Validators.Length];
+            consensusContext.FutureRecoveryPayloads = new ConsensusPayload[consensusContext.Validators.Length];
+
             var copiedContext = TestUtils.CopyMsgBySerialization(consensusContext, new ConsensusContext(null, null));
 
             copiedContext.Block.PrevHash.Should().Be(consensusContext.Block.PrevHash);
@@ -225,6 +230,10 @@ public void TestSerializeAndDeserializeConsensusContext()
             copiedContext.PreparationPayloads.Should().BeEquivalentTo(consensusContext.PreparationPayloads);
             copiedContext.CommitPayloads.Should().BeEquivalentTo(consensusContext.CommitPayloads);
             copiedContext.ChangeViewPayloads.Should().BeEquivalentTo(consensusContext.ChangeViewPayloads);
+            copiedContext.FutureCommitPayloads.Should().BeEquivalentTo(consensusContext.FutureCommitPayloads);
+            copiedContext.FuturePreparationPayloads.Should().BeEquivalentTo(consensusContext.FuturePreparationPayloads);
+            copiedContext.FutureChangeViewPayloads.Should().BeEquivalentTo(consensusContext.FutureChangeViewPayloads);
+            copiedContext.FutureRecoveryPayloads.Should().BeEquivalentTo(consensusContext.FutureRecoveryPayloads);
         }
 
         [TestMethod]

neo/Consensus/ConsensusContext.cs

@@ -37,6 +37,16 @@ internal class ConsensusContext : IDisposable, ISerializable
         // if this node never heard from validator i, LastSeenMessage[i] will be -1.
         public int[] LastSeenMessage;
 
+        /// <summary>
+        /// Future payloads used when CN are probably lagged
+        /// Verified ConsensusPayloads are cached and the lastest height are stored
+        /// They are possibly used when the nodes syncs to the last height
+        /// </summary>
+        public ConsensusPayload[] FuturePreparationPayloads;
+        public ConsensusPayload[] FutureCommitPayloads;
+        public ConsensusPayload[] FutureChangeViewPayloads;
+        public ConsensusPayload[] FutureRecoveryPayloads;
+
         /// <summary>
         /// Store all verified unsorted transactions' senders' fee currently in the consensus context.
         /// </summary>
@@ -107,12 +117,20 @@ public void Deserialize(BinaryReader reader)
                 Block.NextConsensus = null;
             Block.ConsensusData = reader.ReadSerializable<ConsensusData>();
             ViewNumber = reader.ReadByte();
+
             TransactionHashes = reader.ReadSerializableArray<UInt256>();
             Transaction[] transactions = reader.ReadSerializableArray<Transaction>(Block.MaxTransactionsPerBlock);
+
             PreparationPayloads = reader.ReadNullableArray<ConsensusPayload>(Blockchain.MaxValidators);
             CommitPayloads = reader.ReadNullableArray<ConsensusPayload>(Blockchain.MaxValidators);
             ChangeViewPayloads = reader.ReadNullableArray<ConsensusPayload>(Blockchain.MaxValidators);
             LastChangeViewPayloads = reader.ReadNullableArray<ConsensusPayload>(Blockchain.MaxValidators);
+
+            FuturePreparationPayloads = reader.ReadNullableArray<ConsensusPayload>(Blockchain.MaxValidators);
+            FutureCommitPayloads = reader.ReadNullableArray<ConsensusPayload>(Blockchain.MaxValidators);
+            FutureChangeViewPayloads = reader.ReadNullableArray<ConsensusPayload>(Blockchain.MaxValidators);
+            FutureRecoveryPayloads = reader.ReadNullableArray<ConsensusPayload>(Blockchain.MaxValidators);
+
             if (TransactionHashes.Length == 0 && !RequestSentOrReceived)
                 TransactionHashes = null;
             Transactions = transactions.Length == 0 && !RequestSentOrReceived ? null : transactions.ToDictionary(p => p.Hash);
@@ -336,7 +354,7 @@ public ConsensusPayload MakePrepareResponse()
             });
         }
 
-        public void Reset(byte viewNumber)
+        public void Reset(byte viewNumber, bool resetFutures = true)
         {
             if (viewNumber == 0)
             {
@@ -405,6 +423,38 @@ public void Reset(byte viewNumber)
             TransactionHashes = null;
             PreparationPayloads = new ConsensusPayload[Validators.Length];
             if (MyIndex >= 0) LastSeenMessage[MyIndex] = (int)Block.Index;
+
+            if (resetFutures)
+                ResetFuturePayloads();
+        }
+
+        public void ResetFuturePayloads()
+        {
+            FutureCommitPayloads = new ConsensusPayload[Validators.Length];
+            FuturePreparationPayloads = new ConsensusPayload[Validators.Length];
+            FutureChangeViewPayloads = new ConsensusPayload[Validators.Length];
+            FutureRecoveryPayloads = new ConsensusPayload[Validators.Length];
+        }
+
+        public bool HasFuturePayloads()
+        {
+            if (FuturePreparationPayloads != null)
+                foreach (var payload in FuturePreparationPayloads)
+                    if (payload != null)
+                        return true;
+            if (FutureChangeViewPayloads != null)
+                foreach (var payload in FutureChangeViewPayloads)
+                    if (payload != null)
+                        return true;
+            if (FutureCommitPayloads != null)
+                foreach (var payload in FutureCommitPayloads)
+                    if (payload != null)
+                        return true;
+            if (FutureRecoveryPayloads != null)
+                foreach (var payload in FutureRecoveryPayloads)
+                    if (payload != null)
+                        return true;
+            return false;
         }
 
         public void Save()
@@ -422,10 +472,17 @@ public void Serialize(BinaryWriter writer)
             writer.Write(ViewNumber);
             writer.Write(TransactionHashes ?? new UInt256[0]);
             writer.Write(Transactions?.Values.ToArray() ?? new Transaction[0]);
+
             writer.WriteNullableArray(PreparationPayloads);
             writer.WriteNullableArray(CommitPayloads);
             writer.WriteNullableArray(ChangeViewPayloads);
             writer.WriteNullableArray(LastChangeViewPayloads);
+
+            writer.WriteNullableArray(FuturePreparationPayloads);
+            writer.WriteNullableArray(FutureCommitPayloads);
+            writer.WriteNullableArray(FutureChangeViewPayloads);
+            writer.WriteNullableArray(FutureRecoveryPayloads);
+
         }
     }
 }

neo/Consensus/ConsensusService.cs

@@ -161,7 +161,8 @@ private void CheckPreparations()
 
         private void InitializeConsensus(byte viewNumber)
         {
-            context.Reset(viewNumber);
+            bool hasFuturePayload = context.HasFuturePayloads();
+            context.Reset(viewNumber, !hasFuturePayload);
             if (viewNumber > 0)
                 Log($"changeview: view={viewNumber} primary={context.Validators[context.GetPrimaryIndex((byte)(viewNumber - 1u))]}", LogLevel.Warning);
             Log($"initialize: height={context.Block.Index} view={viewNumber} index={context.MyIndex} role={(context.IsPrimary ? "Primary" : context.WatchOnly ? "WatchOnly" : "Backup")}");
@@ -185,8 +186,36 @@ private void InitializeConsensus(byte viewNumber)
             {
                 ChangeTimer(TimeSpan.FromMilliseconds(Blockchain.MillisecondsPerBlock << (viewNumber + 1)));
             }
+
+            // Try to speed up consensus if cached future payloads exists
+            if (hasFuturePayload)
+            {
+                TryToConsumeFuturePayload(context.FuturePreparationPayloads);
+                TryToConsumeFuturePayload(context.FutureCommitPayloads);
+                TryToConsumeFuturePayload(context.FutureChangeViewPayloads);
+                TryToConsumeFuturePayload(context.FutureRecoveryPayloads);
+            }
         }
 
+        private void TryToConsumeFuturePayload(ConsensusPayload[] payloadsArray)
+        {
+            for (int p = 0; p < payloadsArray.Length; p++)
+            {
+                var payload = payloadsArray[p];
+                if (payload != null)
+                {
+                    if (payload.BlockIndex < context.Block.Index)
+                    {
+                        payloadsArray[p] = null;
+                    }
+                    else if (payload.PrevHash == context.Block.PrevHash && payload.BlockIndex == context.Block.Index)
+                    {
+                        OnConsensusPayload(payload, true);
+                        payloadsArray[p] = null;
+                    }
+                }
+            }
+        }
         private void Log(string message, LogLevel level = LogLevel.Info)
         {
             Plugin.Log(nameof(ConsensusService), level, message);
@@ -252,10 +281,74 @@ private void ExtendTimerByFactor(int maxDelayInBlockTimes)
                 ChangeTimer(nextDelay);
         }
 
-        private void OnConsensusPayload(ConsensusPayload payload)
+        private void TryToSaveFuturePayloads(ConsensusPayload payload)
+        {
+            // Limit the maximum future payload to be saved, 
+            // avoiding caching payload when node is lagged behind by more than furtherPayloadToSave locks
+            uint furtherPayloadToSave = 10;
+            if ((payload.BlockIndex >= context.Block.Index + 1) && (payload.BlockIndex <= context.Block.Index + furtherPayloadToSave))
+            {
+                ConsensusMessage futureMessage;
+                try
+                {
+                    futureMessage = payload.ConsensusMessage;
+                }
+                catch (FormatException)
+                {
+                    return;
+                }
+                catch (IOException)
+                {
+                    return;
+                }
+
+                switch (futureMessage)
+                {
+                    case ChangeView view:
+                        tryToSavePayloadIntoArray(context.FutureChangeViewPayloads, payload, futureMessage.ViewNumber);
+                        break;
+                    case Commit commit:
+                        tryToSavePayloadIntoArray(context.FutureCommitPayloads, payload, futureMessage.ViewNumber);
+                        break;
+                    case RecoveryMessage recovery:
+                        tryToSavePayloadIntoArray(context.FutureRecoveryPayloads, payload, futureMessage.ViewNumber);
+                        break;
+                    case PrepareRequest request:
+                    case PrepareResponse response:
+                        tryToSavePayloadIntoArray(context.FuturePreparationPayloads, payload, futureMessage.ViewNumber);
+                        break;
+                }
+            }
+        }
+
+        private void tryToSavePayloadIntoArray(ConsensusPayload[] payloadsArray, ConsensusPayload payload, byte pViewNumber)
+        {
+            byte lastViewNumber = payloadsArray[payload.ValidatorIndex] != null ? (byte)payloadsArray[payload.ValidatorIndex]?.GetDeserializedMessage<ChangeView>().ViewNumber : (byte)0;
+            uint lastHeight = payloadsArray[payload.ValidatorIndex] != null ? payloadsArray[payload.ValidatorIndex].BlockIndex : 0;
+            if (payload.BlockIndex < lastHeight)
+            {
+                Log($"Trying to save validator {payload.ValidatorIndex} payload from height {payload.BlockIndex} but payload of {lastHeight} is already known", LogLevel.Warning);
+            }
+            else if (payload.BlockIndex == lastHeight)
+            {
+                payloadsArray[payload.ValidatorIndex] = (lastViewNumber != 0 && pViewNumber > lastViewNumber) ? payload : payloadsArray[payload.ValidatorIndex];
+            }
+            else
+            {
+                payloadsArray[payload.ValidatorIndex] = payload;
+            }
+        }
+
+        private void OnConsensusPayload(ConsensusPayload payload, bool avoidSaving = false)
         {
-            if (context.BlockSent) return;
             if (payload.Version != context.Block.Version) return;
+            if (payload.ValidatorIndex >= context.Validators.Length) return;
+
+            // Avoiding saving payload when it is being consumed or processed due to a recover payload
+            if (!avoidSaving)
+                TryToSaveFuturePayloads(payload);
+
+            if (context.BlockSent) return;
             if (payload.PrevHash != context.Block.PrevHash || payload.BlockIndex != context.Block.Index)
             {
                 if (context.Block.Index < payload.BlockIndex)
@@ -264,7 +357,6 @@ private void OnConsensusPayload(ConsensusPayload payload)
                 }
                 return;
             }
-            if (payload.ValidatorIndex >= context.Validators.Length) return;
             ConsensusMessage message;
             try
             {
@@ -629,7 +721,7 @@ private void RequestChangeView(ChangeViewReason reason)
         private bool ReverifyAndProcessPayload(ConsensusPayload payload)
         {
             if (!payload.Verify(context.Snapshot)) return false;
-            OnConsensusPayload(payload);
+            OnConsensusPayload(payload, true);
             return true;
         }