Merge pull request #175 from input-output-hk/feature/fast-sync-bugfix

Fast sync download fix on restart

Author: LukasGasior1

src/main/scala/io/iohk/ethereum/blockchain/sync/FastSync.scala

@@ -149,18 +149,23 @@ trait FastSync {
 
     private var assignedHandlers: Map[ActorRef, ActorRef] = Map.empty
 
-    private val syncStateStorageActor= context.actorOf(Props[FastSyncStateActor], "state-storage")
+    private val syncStateStorageActor = context.actorOf(Props[FastSyncStateActor], "state-storage")
+
+    private var requestedMptNodes: Map[ActorRef, Seq[HashType]] = Map.empty
+    private var requestedNonMptNodes: Map[ActorRef, Seq[HashType]] = Map.empty
+    private var requestedBlockBodies: Map[ActorRef, Seq[ByteString]] = Map.empty
+    private var requestedReceipts: Map[ActorRef, Seq[ByteString]] = Map.empty
 
     syncStateStorageActor ! fastSyncStateStorage
 
     private val syncStatePersistCancellable =
       scheduler.schedule(persistStateSnapshotInterval, persistStateSnapshotInterval) {
         syncStateStorageActor ! SyncState(
           initialSyncState.targetBlock,
-          mptNodesQueue,
-          nonMptNodesQueue,
-          blockBodiesQueue,
-          receiptsQueue,
+          requestedMptNodes.values.flatten.toSeq.distinct ++ mptNodesQueue,
+          requestedNonMptNodes.values.flatten.toSeq.distinct ++ nonMptNodesQueue,
+          requestedBlockBodies.values.flatten.toSeq.distinct ++ blockBodiesQueue,
+          requestedReceipts.values.flatten.toSeq.distinct ++ receiptsQueue,
           downloadedNodesCount,
           bestBlockHeaderNumber)
       }
@@ -195,11 +200,17 @@ trait FastSync {
       case SyncRequestHandler.Done =>
         context unwatch sender()
         assignedHandlers -= sender()
+        cleanupRequestedMaps(sender())
         processSyncing()
 
       case Terminated(ref) if assignedHandlers.contains(ref) =>
         context unwatch ref
         assignedHandlers -= ref
+        mptNodesQueue ++= requestedMptNodes.getOrElse(ref, Nil)
+        nonMptNodesQueue ++= requestedNonMptNodes.getOrElse(ref, Nil)
+        blockBodiesQueue ++= requestedBlockBodies.getOrElse(ref, Nil)
+        receiptsQueue ++= requestedReceipts.getOrElse(ref, Nil)
+        cleanupRequestedMaps(ref)
 
       case PrintStatus =>
         val totalNodesCount = downloadedNodesCount + mptNodesQueue.size + nonMptNodesQueue.size
@@ -209,6 +220,13 @@ trait FastSync {
               |State: $downloadedNodesCount/$totalNodesCount known nodes.""".stripMargin.replace("\n", " "))
     }
 
+    private def cleanupRequestedMaps(handler: ActorRef): Unit = {
+      requestedMptNodes = requestedMptNodes - handler
+      requestedNonMptNodes = requestedNonMptNodes - handler
+      requestedBlockBodies = requestedBlockBodies - handler
+      requestedReceipts = requestedReceipts - handler
+    }
+
     private def insertBlocks(requestedHashes: Seq[ByteString], blockBodies: Seq[BlockBody]): Unit = {
       //todo this is moved from FastSyncBlockBodiesRequestHandler.scala we should add block validation here
       //load header from chain by hash and check consistency with BlockValidator.validateHeaderAndBody
@@ -300,6 +318,7 @@ trait FastSync {
       context watch handler
       assignedHandlers += (handler -> peer)
       receiptsQueue = remainingReceipts
+      requestedReceipts += handler -> receiptsToGet
     }
 
     def requestBlockBodies(peer: ActorRef): Unit = {
@@ -308,6 +327,7 @@ trait FastSync {
       context watch handler
       assignedHandlers += (handler -> peer)
       blockBodiesQueue = remainingBlockBodies
+      requestedBlockBodies += handler -> blockBodiesToGet
     }
 
     def requestBlockHeaders(peer: ActorRef): Unit = {
@@ -331,6 +351,8 @@ trait FastSync {
       assignedHandlers += (handler -> peer)
       nonMptNodesQueue = remainingNonMptNodes
       mptNodesQueue = remainingMptNodes
+      requestedMptNodes += handler -> mptNodesToGet
+      requestedNonMptNodes += handler -> nonMptNodesToGet
     }
 
     def unassignedPeers: Set[ActorRef] = peersToDownloadFrom.keySet diff assignedHandlers.values.toSet