Restore MemoryPool transactions from saved consensus context.

neo.UnitTests/TestBlockchain.cs

@@ -4,6 +4,7 @@
 using Neo.Ledger;
 using Neo.Persistence;
 using System;
+using Neo.Persistence.LevelDB;
 
 namespace Neo.UnitTests
 {
@@ -32,6 +33,7 @@ public static NeoSystem InitializeMockNeoSystem()
                 mockSnapshot.SetupGet(p => p.HeaderHashIndex).Returns(new TestMetaDataCache<HashIndexState>());
 
                 var mockStore = new Mock<Store>();
+                mockStore.Setup(p => p.Get(Prefixes.CN_Context, It.IsAny<byte[]>())).Returns((byte[]) null);
 
                 var defaultTx = TestUtils.CreateRandomHashInvocationMockTransaction().Object;
                 mockStore.Setup(p => p.GetBlocks()).Returns(new TestDataCache<UInt256, BlockState>());

neo.UnitTests/UT_Consensus.cs

@@ -5,7 +5,6 @@
 using Moq;
 using Neo.Consensus;
 using Neo.Cryptography;
-using Neo.IO;
 using Neo.Ledger;
 using Neo.Network.P2P;
 using Neo.Network.P2P.Payloads;
@@ -16,6 +15,7 @@
 using System.Linq;
 using System.Numerics;
 using System.Security.Cryptography;
+using Neo.Persistence.LevelDB;
 using ECPoint = Neo.Cryptography.ECC.ECPoint;
 
 namespace Neo.UnitTests

neo/Consensus/ConsensusService.cs

@@ -11,7 +11,6 @@
 using Neo.Wallets;
 using System;
 using System.Collections.Generic;
-using System.IO;
 using System.Linq;
 
 namespace Neo.Consensus
@@ -22,8 +21,6 @@ public class Start { }
         public class SetViewNumber { public byte ViewNumber; }
         internal class Timer { public uint Height; public byte ViewNumber; }
 
-        private const byte ContextSerializationPrefix = 0xf4;
-
         private readonly IConsensusContext context;
         private readonly IActorRef localNode;
         private readonly IActorRef taskManager;
@@ -122,7 +119,7 @@ private void CheckPreparations()
                 ConsensusPayload payload = context.MakeCommit();
                 Log($"send commit");
                 context.State |= ConsensusState.CommitSent;
-                store.Put(ContextSerializationPrefix, new byte[0], context.ToArray());
+                context.WriteContextToStore(store);
                 localNode.Tell(new LocalNode.SendDirectly { Inventory = payload });
                 // Set timer, so we will resend the commit in case of a networking issue
                 ChangeTimer(TimeSpan.FromSeconds(Blockchain.SecondsPerBlock));
@@ -272,6 +269,38 @@ private void OnPersistCompleted(Block block)
             InitializeConsensus(0);
         }
 
+        private void ObtainTransactionsForConsensus()
+        {
+            Dictionary<UInt256, Transaction> mempoolVerified =
+                Blockchain.Singleton.MemPool.GetVerifiedTransactions().ToDictionary(p => p.Hash);
+            List<Transaction> unverified = new List<Transaction>();
+            foreach (UInt256 hash in context.TransactionHashes.Skip(1))
+            {
+                if (mempoolVerified.TryGetValue(hash, out Transaction tx))
+                {
+                    if (!AddTransaction(tx, false))
+                        return;
+                }
+                else
+                {
+                    if (Blockchain.Singleton.MemPool.TryGetValue(hash, out tx))
+                        unverified.Add(tx);
+                }
+            }
+
+            foreach (Transaction tx in unverified)
+                if (!AddTransaction(tx, true))
+                    return;
+            if (context.Transactions.Count < context.TransactionHashes.Length)
+            {
+                UInt256[] hashes = context.TransactionHashes.Where(i => !context.Transactions.ContainsKey(i)).ToArray();
+                taskManager.Tell(new TaskManager.RestartTasks
+                {
+                    Payload = InvPayload.Create(InventoryType.TX, hashes)
+                });
+            }
+        }
+
         private void OnRecoveryMessageReceived(ConsensusPayload payload, RecoveryMessage message)
         {
             if (message.ViewNumber < context.ViewNumber) return;
@@ -329,38 +358,15 @@ private void OnPrepareRequestReceived(ConsensusPayload payload, PrepareRequest m
                         context.PreparationPayloads[i] = null;
             context.PreparationPayloads[payload.ValidatorIndex] = payload;
             byte[] hashData = context.MakeHeader().GetHashData();
+
             for (int i = 0; i < context.CommitPayloads.Length; i++)
                 if (context.CommitPayloads[i] != null)
                     if (!Crypto.Default.VerifySignature(hashData, context.CommitPayloads[i].GetDeserializedMessage<Commit>().Signature, context.Validators[i].EncodePoint(false)))
                         context.CommitPayloads[i] = null;
-            Dictionary<UInt256, Transaction> mempoolVerified = Blockchain.Singleton.MemPool.GetVerifiedTransactions().ToDictionary(p => p.Hash);
 
-            List<Transaction> unverified = new List<Transaction>();
-            foreach (UInt256 hash in context.TransactionHashes.Skip(1))
-            {
-                if (mempoolVerified.TryGetValue(hash, out Transaction tx))
-                {
-                    if (!AddTransaction(tx, false))
-                        return;
-                }
-                else
-                {
-                    if (Blockchain.Singleton.MemPool.TryGetValue(hash, out tx))
-                        unverified.Add(tx);
-                }
-            }
-            foreach (Transaction tx in unverified)
-                if (!AddTransaction(tx, true))
-                    return;
             if (!AddTransaction(message.MinerTransaction, true)) return;
-            if (context.Transactions.Count < context.TransactionHashes.Length)
-            {
-                UInt256[] hashes = context.TransactionHashes.Where(i => !context.Transactions.ContainsKey(i)).ToArray();
-                taskManager.Tell(new TaskManager.RestartTasks
-                {
-                    Payload = InvPayload.Create(InventoryType.TX, hashes)
-                });
-            }
+
+            ObtainTransactionsForConsensus();
         }
 
         private void OnPrepareResponseReceived(ConsensusPayload payload, PrepareResponse message)
@@ -412,24 +418,17 @@ private void OnStart()
         {
             Log("OnStart");
             started = true;
-            byte[] data = store.Get(ContextSerializationPrefix, new byte[0]);
-            if (data != null)
+            bool loadedState = context.LoadContextFromStore(store);
+            if (loadedState && context.State.HasFlag(ConsensusState.CommitSent) && Blockchain.Singleton.Height + 1 == context.BlockIndex)
             {
-                using (MemoryStream ms = new MemoryStream(data, false))
-                using (BinaryReader reader = new BinaryReader(ms))
-                {
-                    context.Deserialize(reader);
-                }
-            }
-            if (context.State.HasFlag(ConsensusState.CommitSent) && context.BlockIndex == Blockchain.Singleton.Height + 1)
                 CheckPreparations();
-            else
-            {
-                InitializeConsensus(0);
-                // Issue a ChangeView with NewViewNumber of 0 to request recovery messages on start-up.
-                if (context.BlockIndex == Blockchain.Singleton.HeaderHeight + 1)
-                    localNode.Tell(new LocalNode.SendDirectly { Inventory = context.MakeChangeView(0) });
+                return;
             }
+
+            InitializeConsensus(0);
+            // Issue a ChangeView with NewViewNumber of 0 to request recovery messages on start-up.
+            if (context.BlockIndex == Blockchain.Singleton.HeaderHeight + 1)
+                localNode.Tell(new LocalNode.SendDirectly { Inventory = context.MakeChangeView(0) });
         }
 
         private void OnTimer(Timer timer)

neo/Consensus/Helper.cs

@@ -0,0 +1,44 @@
+using System.IO;
+using Neo.IO;
+using Neo.Ledger;
+using Neo.Persistence;
+using Neo.Persistence.LevelDB;
+
+namespace Neo.Consensus
+{
+    public static class Helper
+    {
+        internal static void WriteContextToStore(this IConsensusContext context, Store store)
+        {
+            store.Put(Prefixes.CN_Context, new byte[0], context.ToArray());
+        }
+
+        internal static bool LoadContextFromStore(this IConsensusContext context, Store store, bool shouldReset=true)
+        {
+            byte[] data = store.Get(Prefixes.CN_Context, new byte[0]);
+            if (data != null)
+            {
+                if (shouldReset) context.Reset(0);
+                using (MemoryStream ms = new MemoryStream(data, false))
+                using (BinaryReader reader = new BinaryReader(ms))
+                {
+                    context.Deserialize(reader);
+                    return true;
+                }
+            }
+            return false;
+        }
+
+        internal static void LoadTransactionsToMemoryPoolFromSavedConsensusContext(MemoryPool memoryPool, Store store, Store consensusStore)
+        {
+            IConsensusContext context = new ConsensusContext(null);
+            context.LoadContextFromStore(consensusStore, false);
+            if (context.Transactions == null) return;
+            foreach (var tx in context.Transactions.Values)
+            {
+                if (store.ContainsTransaction(tx.Hash)) continue;
+                memoryPool.TryAdd(tx.Hash, tx);
+            }
+        }
+    }
+}

neo/Ledger/Blockchain.cs

@@ -145,11 +145,13 @@ static Blockchain()
             GenesisBlock.RebuildMerkleRoot();
         }
 
-        public Blockchain(NeoSystem system, Store store)
+        public Blockchain(NeoSystem system, Store store, Store consensusStore)
         {
             this.system = system;
-            this.MemPool = new MemoryPool(system, MemoryPoolMaxTransactions);
             this.Store = store;
+            this.MemPool = new MemoryPool(system, MemoryPoolMaxTransactions);
+            Consensus.Helper.LoadTransactionsToMemoryPoolFromSavedConsensusContext(MemPool, store, consensusStore);
+
             lock (lockObj)
             {
                 if (singleton != null)
@@ -683,9 +685,9 @@ internal static void ProcessValidatorStateDescriptor(StateDescriptor descriptor,
             }
         }
 
-        public static Props Props(NeoSystem system, Store store)
+        public static Props Props(NeoSystem system, Store store, Store consensusStore)
         {
-            return Akka.Actor.Props.Create(() => new Blockchain(system, store)).WithMailbox("blockchain-mailbox");
+            return Akka.Actor.Props.Create(() => new Blockchain(system, store, consensusStore)).WithMailbox("blockchain-mailbox");
         }
 
         private void SaveHeaderHashList(Snapshot snapshot = null)

neo/NeoSystem.cs

@@ -26,14 +26,14 @@ public class NeoSystem : IDisposable
         public IActorRef Consensus { get; private set; }
         public RpcServer RpcServer { get; private set; }
 
-        private readonly Store store;
+        private readonly Store consensusStore;
         private Peer.Start start_message = null;
         private bool suspend = false;
 
-        public NeoSystem(Store store)
+        public NeoSystem(Store store, Store consensusStore = null)
         {
-            this.store = store;
-            this.Blockchain = ActorSystem.ActorOf(Ledger.Blockchain.Props(this, store));
+            this.consensusStore = consensusStore ?? store;
+            this.Blockchain = ActorSystem.ActorOf(Ledger.Blockchain.Props(this, store, this.consensusStore));
             this.LocalNode = ActorSystem.ActorOf(Network.P2P.LocalNode.Props(this));
             this.TaskManager = ActorSystem.ActorOf(Network.P2P.TaskManager.Props(this));
             Plugin.LoadPlugins(this);
@@ -56,9 +56,9 @@ internal void ResumeNodeStartup()
             }
         }
 
-        public void StartConsensus(Wallet wallet, Store consensus_store = null)
+        public void StartConsensus(Wallet wallet)
         {
-            Consensus = ActorSystem.ActorOf(ConsensusService.Props(this.LocalNode, this.TaskManager, consensus_store ?? store, wallet));
+            Consensus = ActorSystem.ActorOf(ConsensusService.Props(this.LocalNode, this.TaskManager, consensusStore, wallet));
             Consensus.Tell(new ConsensusService.Start());
         }
 

neo/Persistence/LevelDB/Prefixes.cs

@@ -19,5 +19,6 @@ internal static class Prefixes
         public const byte IX_CurrentHeader = 0xc1;
 
         public const byte SYS_Version = 0xf0;
+        public const byte CN_Context = 0xf4;
     }
 }