Restore MemoryPool transactions from saved consensus context.

neo.UnitTests/TestBlockchain.cs

@@ -32,6 +32,7 @@ public static NeoSystem InitializeMockNeoSystem()
                 mockSnapshot.SetupGet(p => p.HeaderHashIndex).Returns(new TestMetaDataCache<HashIndexState>());
 
                 var mockStore = new Mock<Store>();
+                mockStore.Setup(p => p.Get(Consensus.Helper.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;

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 bool IgnoreRecoveryLogs; }
         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;
@@ -127,7 +124,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));
@@ -412,27 +409,17 @@ private void OnStart(Start options)
         {
             Log("OnStart");
             started = true;
-            if (!options.IgnoreRecoveryLogs)
+            bool loadedState = !options.IgnoreRecoveryLogs && context.LoadContextFromStore(store);
+            if (loadedState && context.State.HasFlag(ConsensusState.CommitSent) && Blockchain.Singleton.Height + 1 == context.BlockIndex)
             {
-                byte[] data = store.Get(ContextSerializationPrefix, new byte[0]);
-                if (data != null)
-                {
-                    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 Neo.IO;
+using Neo.Network.P2P.Payloads;
+using Neo.Persistence;
+using System.Collections.Generic;
+using System.IO;
+
+namespace Neo.Consensus
+{
+    internal static class Helper
+    {
+        /// <summary>
+        /// Prefix for saving consensus state.
+        /// </summary>
+        public const byte CN_Context = 0xf4;
+
+        public static void WriteContextToStore(this IConsensusContext context, Store store)
+        {
+            store.PutSync(CN_Context, new byte[0], context.ToArray());
+        }
+
+        public static bool LoadContextFromStore(this IConsensusContext context, Store store, bool shouldReset = true)
+        {
+            byte[] data = store.Get(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;
+        }
+
+        public static IEnumerable<Transaction> RetreiveTransactionsFromSavedConsensusContext(Store consensusStore)
+        {
+            IConsensusContext context = new ConsensusContext(null);
+            context.LoadContextFromStore(consensusStore, false);
+            return context.Transactions?.Values ?? (IEnumerable<Transaction>)new Transaction[0];
+        }
+    }
+}

neo/Ledger/Blockchain.cs

@@ -148,8 +148,9 @@ static Blockchain()
         public Blockchain(NeoSystem system, Store store)
         {
             this.system = system;
-            this.MemPool = new MemoryPool(system, MemoryPoolMaxTransactions);
             this.Store = store;
+            this.MemPool = new MemoryPool(system, MemoryPoolMaxTransactions);
+
             lock (lockObj)
             {
                 if (singleton != null)

neo/NeoSystem.cs

@@ -26,13 +26,15 @@ public class NeoSystem : IDisposable
         public IActorRef Consensus { get; private set; }
         public RpcServer RpcServer { get; private set; }
 
-        private readonly Store store;
+        private readonly Store blockchainStore;
+        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.blockchainStore = store;
+            this.consensusStore = consensusStore ?? store;
             this.Blockchain = ActorSystem.ActorOf(Ledger.Blockchain.Props(this, store));
             this.LocalNode = ActorSystem.ActorOf(Network.P2P.LocalNode.Props(this));
             this.TaskManager = ActorSystem.ActorOf(Network.P2P.TaskManager.Props(this));
@@ -56,9 +58,9 @@ internal void ResumeNodeStartup()
             }
         }
 
-        public void StartConsensus(Wallet wallet, Store consensus_store = null, bool ignoreRecoveryLogs = false)
+        public void StartConsensus(Wallet wallet, bool ignoreRecoveryLogs = false)
         {
-            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 { IgnoreRecoveryLogs = ignoreRecoveryLogs });
         }
 
@@ -74,6 +76,10 @@ public void StartConsensus(Wallet wallet, Store consensus_store = null, bool ign
             };
             if (!suspend)
             {
+                var consensusTransactions = Neo.Consensus.Helper.RetreiveTransactionsFromSavedConsensusContext(consensusStore);
+                foreach (var tx in consensusTransactions)
+                    Blockchain.Tell(tx, ActorRefs.NoSender);
+
                 LocalNode.Tell(start_message);
                 start_message = null;
             }

neo/Persistence/LevelDB/LevelDBStore.cs

@@ -116,5 +116,10 @@ public override void Put(byte prefix, byte[] key, byte[] value)
         {
             db.Put(WriteOptions.Default, SliceBuilder.Begin(prefix).Add(key), value);
         }
+
+        public override void PutSync(byte prefix, byte[] key, byte[] value)
+        {
+            db.Put(new WriteOptions { Sync = true }, SliceBuilder.Begin(prefix).Add(key), value);
+        }
     }
 }

neo/Persistence/LevelDB/Prefixes.cs

@@ -19,5 +19,10 @@ internal static class Prefixes
         public const byte IX_CurrentHeader = 0xc1;
 
         public const byte SYS_Version = 0xf0;
+
+        /* Prefixes 0xf1 to 0xff are reserved for external use.
+         *
+         * Note: The saved consensus state uses the Prefix 0xf4
+         */
     }
 }

neo/Persistence/Store.cs

@@ -36,6 +36,7 @@ public abstract class Store : IPersistence
         public abstract MetaDataCache<HashIndexState> GetBlockHashIndex();
         public abstract MetaDataCache<HashIndexState> GetHeaderHashIndex();
         public abstract void Put(byte prefix, byte[] key, byte[] value);
+        public abstract void PutSync(byte prefix, byte[] key, byte[] value);
 
         public abstract Snapshot GetSnapshot();
     }