Add header back

src/neo/Ledger/Blockchain.cs

@@ -72,6 +72,7 @@ public class RelayResult { public IInventory Inventory; public VerifyResult Resu
         public DataCache View => new SnapshotCache(Store);
         public MemoryPool MemPool { get; }
         public uint Height => NativeContract.Ledger.CurrentIndex(currentSnapshot);
+        public uint HeaderHeight => NativeContract.Ledger.CurrentHeaderIndex(currentSnapshot);
         public UInt256 CurrentBlockHash => NativeContract.Ledger.CurrentHash(currentSnapshot);
 
         private static Blockchain singleton;
@@ -216,30 +217,83 @@ private VerifyResult OnNewBlock(Block block)
         {
             if (block.Index <= Height)
                 return VerifyResult.AlreadyExists;
-            if (block.Index - 1 > Height)
+            if (block.Index - 1 > HeaderHeight)
             {
                 AddUnverifiedBlockToCache(block);
                 return VerifyResult.UnableToVerify;
             }
-            if (block.Index == Height + 1)
+            if (block.Index == HeaderHeight + 1)
             {
                 if (!block.Verify(currentSnapshot))
                     return VerifyResult.Invalid;
-                block_cache.TryAdd(block.Hash, block);
-                block_cache_unverified.Remove(block.Index);
-                Persist(block);
+            }
+            else
+            {
+                if (!block.Hash.Equals(NativeContract.Ledger.GetBlockHash(currentSnapshot, block.Index)))
+                    return VerifyResult.Invalid;
+            }
+            block_cache.TryAdd(block.Hash, block);
+            if (block.Index == Height + 1)
+            {
+                Block block_persist = block;
+                List<Block> blocksToPersistList = new List<Block>();
+                while (true)
+                {
+                    blocksToPersistList.Add(block_persist);
+                    if (block_persist.Index + 1 >= NativeContract.Ledger.CurrentHeaderIndex(currentSnapshot)) break;
+                    UInt256 hash = NativeContract.Ledger.CurrentHeaderHash(currentSnapshot);
+                    if (!block_cache.TryGetValue(hash, out block_persist)) break;
+                }
+
+                int blocksPersisted = 0;
+                foreach (Block blockToPersist in blocksToPersistList)
+                {
+                    block_cache_unverified.Remove(blockToPersist.Index);
+                    Persist(blockToPersist);
+
+                    // 15000 is the default among of seconds per block, while MilliSecondsPerBlock is the current
+                    uint extraBlocks = (15000 - MillisecondsPerBlock) / 1000;
+
+                    if (blocksPersisted++ < blocksToPersistList.Count - (2 + Math.Max(0, extraBlocks))) continue;
+                    // Empirically calibrated for relaying the most recent 2 blocks persisted with 15s network
+                    // Increase in the rate of 1 block per second in configurations with faster blocks
+
+                    if (blockToPersist.Index + 100 >= NativeContract.Ledger.CurrentHeaderIndex(currentSnapshot))
+                        system.LocalNode.Tell(new LocalNode.RelayDirectly { Inventory = blockToPersist });
+                }
                 if (block_cache_unverified.TryGetValue(Height + 1, out var unverifiedBlocks))
                 {
                     foreach (var unverifiedBlock in unverifiedBlocks.Blocks)
                         Self.Tell(unverifiedBlock, ActorRefs.NoSender);
                     block_cache_unverified.Remove(Height + 1);
                 }
-                // We can store the new block in block_cache and tell the new height to other nodes after Persist().
-                system.LocalNode.Tell(Message.Create(MessageCommand.Ping, PingPayload.Create(Singleton.Height)));
+            }
+            else
+            {
+                if (block.Index + 100 >= NativeContract.Ledger.CurrentHeaderIndex(currentSnapshot))
+                    system.LocalNode.Tell(new LocalNode.RelayDirectly { Inventory = block });
             }
             return VerifyResult.Succeed;
         }
 
+        private void OnNewHeaders(Header[] headers)
+        {
+            using (SnapshotCache snapshot = GetSnapshot())
+            {
+                foreach (Header header in headers)
+                {
+                    if (header.Index - 1 >= HeaderHeight) break;
+                    if (header.Index < HeaderHeight) continue;
+                    if (!header.Verify(snapshot)) break;
+                    NativeContract.Ledger.SaveHeader(snapshot, header);
+                    NativeContract.Ledger.SetCurrentHeader(snapshot, header.Hash, header.Index);
+                }
+                snapshot.Commit();
+            }
+            UpdateCurrentSnapshot();
+            system.TaskManager.Tell(new TaskManager.HeaderTaskCompleted(), Sender);
+        }
+
         private VerifyResult OnNewInventory(IInventory inventory)
         {
             if (!inventory.Verify(currentSnapshot)) return VerifyResult.Invalid;
@@ -271,6 +325,9 @@ protected override void OnReceive(object message)
                 case FillMemoryPool fill:
                     OnFillMemoryPool(fill.Transactions);
                     break;
+                case Header[] headers:
+                    OnNewHeaders(headers);
+                    break;
                 case Block block:
                     OnInventory(block, false);
                     break;
@@ -430,6 +487,7 @@ internal protected override bool IsHighPriority(object message)
         {
             switch (message)
             {
+                case Header[] _:
                 case Block _:
                 case ExtensiblePayload _:
                 case Terminated _:

src/neo/Network/P2P/MessageCommand.cs

@@ -14,10 +14,6 @@ public enum MessageCommand : byte
         GetAddr = 0x10,
         [ReflectionCache(typeof(AddrPayload))]
         Addr = 0x11,
-        [ReflectionCache(typeof(PingPayload))]
-        Ping = 0x18,
-        [ReflectionCache(typeof(PingPayload))]
-        Pong = 0x19,
 
         //synchronization
         [ReflectionCache(typeof(GetBlockByIndexPayload))]

src/neo/Network/P2P/Payloads/Header.cs

@@ -1,3 +1,4 @@
+using Neo.SmartContract.Native;
 using System;
 using System.IO;
 
@@ -35,5 +36,20 @@ public override void Serialize(BinaryWriter writer)
             base.Serialize(writer);
             writer.Write((byte)0);
         }
+
+        public TrimmedBlock Trim()
+        {
+            return new TrimmedBlock
+            {
+                Version = Version,
+                PrevHash = PrevHash,
+                MerkleRoot = MerkleRoot,
+                Timestamp = Timestamp,
+                Index = Index,
+                NextConsensus = NextConsensus,
+                Witness = Witness,
+                Hashes = new UInt256[0]
+            };
+        }
     }
 }

src/neo/Network/P2P/RemoteNode.ProtocolHandler.cs

@@ -90,18 +90,15 @@ private void OnMessage(Message msg)
                 case MessageCommand.GetHeaders:
                     OnGetHeadersMessageReceived((GetBlockByIndexPayload)msg.Payload);
                     break;
+                case MessageCommand.Headers:
+                    OnHeadersMessageReceived((HeadersPayload)msg.Payload);
+                    break;
                 case MessageCommand.Inv:
                     OnInvMessageReceived((InvPayload)msg.Payload);
                     break;
                 case MessageCommand.Mempool:
                     OnMemPoolMessageReceived();
                     break;
-                case MessageCommand.Ping:
-                    OnPingMessageReceived((PingPayload)msg.Payload);
-                    break;
-                case MessageCommand.Pong:
-                    OnPongMessageReceived((PingPayload)msg.Payload);
-                    break;
                 case MessageCommand.Transaction:
                     if (msg.Payload.Size <= Transaction.MaxTransactionSize)
                         OnInventoryReceived((Transaction)msg.Payload);
@@ -110,7 +107,6 @@ private void OnMessage(Message msg)
                 case MessageCommand.Version:
                     throw new ProtocolViolationException();
                 case MessageCommand.Alert:
-                case MessageCommand.Headers:
                 case MessageCommand.MerkleBlock:
                 case MessageCommand.NotFound:
                 case MessageCommand.Reject:
@@ -290,6 +286,12 @@ private void OnGetHeadersMessageReceived(GetBlockByIndexPayload payload)
             EnqueueMessage(Message.Create(MessageCommand.Headers, HeadersPayload.Create(headers.ToArray())));
         }
 
+        private void OnHeadersMessageReceived(HeadersPayload payload)
+        {
+            if (payload.Headers.Length == 0) return;
+            system.Blockchain.Tell(payload.Headers);
+        }
+
         private void OnInventoryReceived(IInventory inventory)
         {
             pendingKnownHashes.Remove(inventory.Hash);
@@ -330,17 +332,6 @@ private void OnMemPoolMessageReceived()
                 EnqueueMessage(Message.Create(MessageCommand.Inv, payload));
         }
 
-        private void OnPingMessageReceived(PingPayload payload)
-        {
-            UpdateLastBlockIndex(payload.LastBlockIndex, true);
-            EnqueueMessage(Message.Create(MessageCommand.Pong, PingPayload.Create(Blockchain.Singleton.Height, payload.Nonce)));
-        }
-
-        private void OnPongMessageReceived(PingPayload payload)
-        {
-            UpdateLastBlockIndex(payload.LastBlockIndex, true);
-        }
-
         private void OnVerackMessageReceived()
         {
             verack = true;

src/neo/Network/P2P/RemoteNode.cs

@@ -77,8 +77,6 @@ private void EnqueueMessage(Message message)
                 case MessageCommand.GetBlocks:
                 case MessageCommand.GetHeaders:
                 case MessageCommand.Mempool:
-                case MessageCommand.Ping:
-                case MessageCommand.Pong:
                     is_single = true;
                     break;
             }
@@ -130,8 +128,6 @@ protected override void OnReceive(object message)
                     {
                         if (payload.LastBlockIndex > LastHeightSent)
                             LastHeightSent = payload.LastBlockIndex;
-                        else if (msg.Command == MessageCommand.Ping)
-                            break;
                     }
                     EnqueueMessage(msg);
                     break;

src/neo/Network/P2P/TaskManager.cs

@@ -19,11 +19,13 @@ public class TaskManager : UntypedActor
         internal class Register { public VersionPayload Version; }
         internal class Update { public uint LastBlockIndex; public bool RequestTasks; }
         internal class NewTasks { public InvPayload Payload; }
+        public class HeaderTaskCompleted { }
         public class RestartTasks { public InvPayload Payload; }
         private class Timer { }
 
         private static readonly TimeSpan TimerInterval = TimeSpan.FromSeconds(30);
         private static readonly TimeSpan TaskTimeout = TimeSpan.FromMinutes(1);
+        private static readonly UInt256 HeaderTaskHash = UInt256.Zero;
         private static readonly UInt256 MemPoolTaskHash = UInt256.Parse("0x0000000000000000000000000000000000000000000000000000000000000001");
 
         private const int MaxConncurrentTasks = 3;
@@ -40,6 +42,7 @@ private class Timer { }
         private readonly HashSet<uint> failedSyncTasks = new HashSet<uint>();
         private readonly Dictionary<IActorRef, TaskSession> sessions = new Dictionary<IActorRef, TaskSession>();
         private readonly ICancelable timer = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(TimerInterval, TimerInterval, Context.Self, new Timer(), ActorRefs.NoSender);
+        private bool HasHeaderTask => globalTasks.ContainsKey(HeaderTaskHash);
         private uint lastTaskIndex = 0;
 
         public TaskManager(NeoSystem system)
@@ -51,6 +54,15 @@ public TaskManager(NeoSystem system)
             Context.System.EventStream.Subscribe(Self, typeof(Blockchain.RelayResult));
         }
 
+        private void OnHeaderTaskCompleted()
+        {
+            if (!sessions.TryGetValue(Sender, out TaskSession session))
+                return;
+            session.InvTasks.Remove(HeaderTaskHash);
+            DecrementGlobalTask(HeaderTaskHash);
+            RequestTasks();
+        }
+
         private bool AssignSyncTask(uint index, TaskSession filterSession = null)
         {
             if (index <= Blockchain.Singleton.Height || sessions.Values.Any(p => p != filterSession && p.IndexTasks.ContainsKey(index)))
@@ -78,7 +90,7 @@ private void OnBlock(Block block)
             if (session is null) return;
             session.IndexTasks.Remove(block.Index);
             receivedBlockIndex.TryAdd(block.Index, session);
-            RequestTasks(false);
+            RequestTasks();
         }
 
         private void OnInvalidBlock(Block invalidBlock)
@@ -121,7 +133,7 @@ private void OnNewTasks(InvPayload payload)
         private void OnPersistCompleted(Block block)
         {
             receivedBlockIndex.Remove(block.Index);
-            RequestTasks(false);
+            RequestTasks();
         }
 
         protected override void OnReceive(object message)
@@ -137,6 +149,9 @@ protected override void OnReceive(object message)
                 case NewTasks tasks:
                     OnNewTasks(tasks.Payload);
                     break;
+                case HeaderTaskCompleted _:
+                    OnHeaderTaskCompleted();
+                    break;
                 case RestartTasks restart:
                     OnRestartTasks(restart.Payload);
                     break;
@@ -169,7 +184,7 @@ private void OnRegister(VersionPayload version)
             if (session.IsFullNode)
                 session.InvTasks.TryAdd(MemPoolTaskHash, TimeProvider.Current.UtcNow);
             sessions.TryAdd(Sender, session);
-            RequestTasks(true);
+            RequestTasks();
         }
 
         private void OnUpdate(Update update)
@@ -178,7 +193,7 @@ private void OnUpdate(Update update)
                 return;
             session.LastBlockIndex = update.LastBlockIndex;
             session.ExpireTime = TimeProvider.Current.UtcNow.AddMilliseconds(PingCoolingOffPeriod);
-            if (update.RequestTasks) RequestTasks(true);
+            if (update.RequestTasks) RequestTasks();
         }
 
         private void OnRestartTasks(InvPayload payload)
@@ -260,7 +275,7 @@ private void OnTimer()
                     }
                 }
             }
-            RequestTasks(true);
+            RequestTasks();
         }
 
         protected override void PostStop()
@@ -274,12 +289,10 @@ public static Props Props(NeoSystem system)
             return Akka.Actor.Props.Create(() => new TaskManager(system)).WithMailbox("task-manager-mailbox");
         }
 
-        private void RequestTasks(bool sendPing)
+        private void RequestTasks()
         {
             if (sessions.Count() == 0) return;
 
-            if (sendPing) SendPingMessage();
-
             while (failedSyncTasks.Count() > 0)
             {
                 var failedTask = failedSyncTasks.First();
@@ -299,33 +312,6 @@ private void RequestTasks(bool sendPing)
                 if (!AssignSyncTask(++lastTaskIndex)) break;
             }
         }
-
-        private void SendPingMessage()
-        {
-            TrimmedBlock block;
-            using (SnapshotCache snapshot = Blockchain.Singleton.GetSnapshot())
-            {
-                block = NativeContract.Ledger.GetTrimmedBlock(snapshot, NativeContract.Ledger.CurrentHash(snapshot));
-            }
-
-            foreach (KeyValuePair<IActorRef, TaskSession> item in sessions)
-            {
-                var node = item.Key;
-                var session = item.Value;
-
-                if (session.ExpireTime < TimeProvider.Current.UtcNow ||
-                     (block.Index >= session.LastBlockIndex &&
-                     TimeProvider.Current.UtcNow.ToTimestampMS() - PingCoolingOffPeriod >= block.Timestamp))
-                {
-                    if (session.InvTasks.Remove(MemPoolTaskHash))
-                    {
-                        node.Tell(Message.Create(MessageCommand.Mempool));
-                    }
-                    node.Tell(Message.Create(MessageCommand.Ping, PingPayload.Create(Blockchain.Singleton.Height)));
-                    session.ExpireTime = TimeProvider.Current.UtcNow.AddMilliseconds(PingCoolingOffPeriod);
-                }
-            }
-        }
     }
 
     internal class TaskManagerMailbox : PriorityMailbox