Restore MemoryPool transactions from saved consensus context.

neo.UnitTests/UT_MemoryPool.cs

@@ -9,6 +9,7 @@
 using Neo.IO.Wrappers;
 using Neo.Network.P2P.Payloads;
 using Neo.Persistence;
+using Neo.Persistence.LevelDB;
 
 namespace Neo.UnitTests
 {
@@ -74,6 +75,7 @@ public void TestSetup()
                 mockStore.Setup(p => p.GetBlockHashIndex()).Returns(new TestMetaDataCache<HashIndexState>());
                 mockStore.Setup(p => p.GetHeaderHashIndex()).Returns(new TestMetaDataCache<HashIndexState>());
                 mockStore.Setup(p => p.GetSnapshot()).Returns(mockSnapshot.Object);
+                mockStore.Setup(p => p.Get(Prefixes.CN_CONTEXT, It.IsAny<byte[]>())).Returns((byte[]) null);
 
                 Console.WriteLine("initialize NeoSystem");
                 TheNeoSystem = new NeoSystem(mockStore.Object); // new Mock<NeoSystem>(mockStore.Object);

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;
@@ -107,6 +104,8 @@ private void CheckExpectedView(byte view_number)
             if (context.ExpectedView.Count(p => p == view_number) >= context.M)
             {
                 InitializeConsensus(view_number);
+                // Save our view so if we crash and come back we will be closer to the correct view.
+                context.WriteContextToStore(store);
             }
         }
 
@@ -117,7 +116,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 });
                 CheckCommits();
             }
@@ -272,7 +271,6 @@ private void OnPrepareRequestReceived(ConsensusPayload payload, PrepareRequest m
                 }
                 else
                 {
-
                     if (Blockchain.Singleton.MemPool.TryGetValue(hash, out tx))
                         unverified.Add(tx);
                 }
@@ -338,19 +336,11 @@ private void OnStart()
         {
             Log("OnStart");
             started = true;
-            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);
-                }
-            }
+            context.LoadContextFromStore(store);
             if (context.State.HasFlag(ConsensusState.CommitSent))
                 CheckPreparations();
             else
-                InitializeConsensus(0);
+                InitializeConsensus(context.ViewNumber);
         }
 
         private void OnTimer(Timer timer)

neo/Consensus/Helper.cs

@@ -0,0 +1,28 @@
+using System.IO;
+using Neo.IO;
+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 void LoadContextFromStore(this IConsensusContext context, Store store)
+        {
+            byte[] data = store.Get(Prefixes.CN_CONTEXT, new byte[0]);
+            if (data != null)
+            {
+                using (MemoryStream ms = new MemoryStream(data, false))
+                using (BinaryReader reader = new BinaryReader(ms))
+                {
+                    context.Deserialize(reader);
+                }
+            }
+        }
+    }
+}

neo/Ledger/MemoryPool.cs

@@ -6,6 +6,7 @@
 using System.Runtime.CompilerServices;
 using System.Threading;
 using Akka.Util.Internal;
+using Neo.Consensus;
 using Neo.Network.P2P;
 using Neo.Persistence;
 using Neo.Plugins;
@@ -107,6 +108,13 @@ public MemoryPool(NeoSystem system, int capacity)
             _system = system;
             Capacity = capacity;
             LoadMaxTxLimitsFromPolicyPlugins();
+            IConsensusContext context = new ConsensusContext(null);
+            context.LoadContextFromStore(system.store);
+            if (context.Transactions == null) return;
+            foreach (var tx in context.Transactions.Values)
+            {
+                TryAdd(tx.Hash, tx);
+            }
         }
 
         public void LoadMaxTxLimitsFromPolicyPlugins()

neo/NeoSystem.cs

@@ -26,7 +26,7 @@ public class NeoSystem : IDisposable
         public IActorRef Consensus { get; private set; }
         public RpcServer RpcServer { get; private set; }
 
-        private readonly Store store;
+        internal readonly Store store;
         private Peer.Start start_message = null;
         private bool suspend = false;
 

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;
     }
 }