Merge remote-tracking branch 'origin/phase/3/blockSync' into feature/privateChain

Author: Lukasz Gasior

src/main/resources/reference.conf

@@ -71,7 +71,7 @@ etc-client {
     max-concurrent-requests = 50
     block-headers-per-request = 2048
     block-bodies-per-request = 128
-    receipts-per-request = 128
+    receipts-per-request = 60
     nodes-per-request = 1000
     min-peers-to-choose-target-block = 2
     target-block-offset = 500

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

@@ -2,20 +2,20 @@ package io.iohk.ethereum.blockchain.sync
 
 import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.ExecutionContext.Implicits.global
-
-import akka.actor.{Scheduler, Cancellable, ActorRef, Actor}
+import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Scheduler}
 
 trait BlacklistSupport {
-  selfActor: Actor =>
+  selfActor: Actor with ActorLogging =>
 
   import BlacklistSupport._
 
   def scheduler: Scheduler
 
   var blacklistedPeers: Seq[(ActorRef, Cancellable)] = Nil
 
-  def blacklist(peer: ActorRef, duration: FiniteDuration): Unit = {
+  def blacklist(peer: ActorRef, duration: FiniteDuration, reason: String): Unit = {
     undoBlacklist(peer)
+    log.info(s"Blacklisting peer (${peer.path.name}), $reason")
     val unblacklistCancellable = scheduler.scheduleOnce(duration, self, UnblacklistPeer(peer))
     blacklistedPeers :+= (peer, unblacklistCancellable)
   }
@@ -30,6 +30,6 @@ trait BlacklistSupport {
 }
 
 object BlacklistSupport {
-  case class BlacklistPeer(peer: ActorRef)
+  case class BlacklistPeer(peer: ActorRef, reason: String)
   case class UnblacklistPeer(peer: ActorRef)
 }

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

@@ -65,15 +65,15 @@ trait FastSync {
         tryStartFastSync(newReceived)
       } else context become waitingForBlockHeaders(newWaitingFor, newReceived, timeout)
 
-    case PeerActor.MessageReceived(BlockHeaders(_)) =>
+    case PeerActor.MessageReceived(BlockHeaders(blockHeaders)) =>
       sender() ! PeerActor.Unsubscribe
-      blacklist(sender(), blacklistDuration)
+      blacklist(sender(), blacklistDuration,s"did not respond with 1 header but with ${blockHeaders.size}, blacklisting for $blacklistDuration")
       context become waitingForBlockHeaders(waitingFor - sender(), received, timeout)
 
     case BlockHeadersTimeout =>
       waitingFor.foreach { peer =>
         peer ! PeerActor.Unsubscribe
-        blacklist(peer, blacklistDuration)
+        blacklist(peer, blacklistDuration, s"did not respond within required time with block header, blacklisting for $blacklistDuration")
       }
       tryStartFastSync(received)
   }
@@ -120,21 +120,13 @@ trait FastSync {
           startFastSync(initialSyncState)
 
         case None =>
-          log.info("Peer ({}) did not respond with target block header, blacklisting and scheduling retry in {}",
-            sender().path.name,
-            startRetryInterval)
-
-          blacklist(sender(), blacklistDuration)
+          blacklist(sender(), blacklistDuration,s"did not respond with target block header, blacklisting and scheduling retry in $startRetryInterval")
           scheduleStartRetry(startRetryInterval)
           context become startingFastSync
       }
 
     case TargetBlockTimeout =>
-      log.info("Peer ({}) did not respond with target block header (timeout), blacklisting and scheduling retry in {}",
-        sender().path.name,
-        startRetryInterval)
-
-      blacklist(sender(), blacklistDuration)
+      blacklist(sender(), blacklistDuration, s"did not respond with target block header (timeout), blacklisting and scheduling retry in $startRetryInterval")
       peer ! PeerActor.Unsubscribe
       scheduleStartRetry(startRetryInterval)
       context become startingFastSync

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

@@ -19,11 +19,12 @@ class FastSyncNodesRequestHandler(
   import FastSyncNodesRequestHandler._
 
   override val requestMsg = GetNodeData(requestedHashes.map(_.v))
-  override val responseMsgCode = NodeData.code
+  override val responseMsgCode: Int = NodeData.code
 
   override def handleResponseMsg(nodeData: NodeData): Unit = {
     if (nodeData.values.isEmpty) {
-      syncController ! BlacklistSupport.BlacklistPeer(peer)
+      val reason = s"got empty mpt node response for known hashes: ${requestedHashes.map(h => Hex.toHexString(h.v.toArray[Byte]))}"
+      syncController ! BlacklistSupport.BlacklistPeer(peer, reason)
     }
 
     val receivedHashes = nodeData.values.map(v => ByteString(kec256(v.toArray[Byte])))
@@ -59,7 +60,8 @@ class FastSyncNodesRequestHandler(
   }
 
   override def handleTimeout(): Unit = {
-    syncController ! BlacklistSupport.BlacklistPeer(peer)
+    val reason = s"time out on mpt node response for known hashes: ${requestedHashes.map(h => Hex.toHexString(h.v.toArray[Byte]))}"
+    syncController ! BlacklistSupport.BlacklistPeer(peer, reason)
     syncController ! FastSync.EnqueueNodes(requestedHashes)
     cleanupAndStop()
   }

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

@@ -5,6 +5,7 @@ import akka.util.ByteString
 import io.iohk.ethereum.db.storage.AppStateStorage
 import io.iohk.ethereum.domain.Blockchain
 import io.iohk.ethereum.network.p2p.messages.PV63.{GetReceipts, Receipts}
+import org.spongycastle.util.encoders.Hex
 
 class FastSyncReceiptsRequestHandler(
     peer: ActorRef,
@@ -14,7 +15,7 @@ class FastSyncReceiptsRequestHandler(
   extends SyncRequestHandler[GetReceipts, Receipts](peer) {
 
   override val requestMsg = GetReceipts(requestedHashes)
-  override val responseMsgCode = Receipts.code
+  override val responseMsgCode: Int = Receipts.code
 
   override def handleResponseMsg(receipts: Receipts): Unit = {
     (requestedHashes zip receipts.receiptsForBlocks).foreach { case (hash, receiptsForBlock) =>
@@ -25,7 +26,8 @@ class FastSyncReceiptsRequestHandler(
     updateBestBlockIfNeeded(receivedHashes)
 
     if (receipts.receiptsForBlocks.isEmpty) {
-      syncController ! BlacklistSupport.BlacklistPeer(peer)
+      val reason = s"got empty receipts for known hashes: ${requestedHashes.map(h => Hex.toHexString(h.toArray[Byte]))}"
+      syncController ! BlacklistSupport.BlacklistPeer(peer, reason)
     }
 
     val remainingReceipts = requestedHashes.drop(receipts.receiptsForBlocks.size)
@@ -54,7 +56,8 @@ class FastSyncReceiptsRequestHandler(
   }
 
   override def handleTimeout(): Unit = {
-    syncController ! BlacklistSupport.BlacklistPeer(peer)
+    val reason = s"time out on receipts response for known hashes: ${requestedHashes.map(h => Hex.toHexString(h.toArray[Byte]))}"
+    syncController ! BlacklistSupport.BlacklistPeer(peer, reason)
     syncController ! FastSync.EnqueueReceipts(requestedHashes)
     cleanupAndStop()
   }

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

@@ -162,7 +162,7 @@ trait RegularSync {
   }
 
   private def resumeWithDifferentPeer(currentPeer: ActorRef) = {
-    self ! BlacklistPeer(currentPeer)
+    self ! BlacklistPeer(currentPeer, "because of error in response")
     headersQueue = Seq.empty
     context.self ! ResumeRegularSync
   }

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

@@ -5,6 +5,7 @@ import akka.util.ByteString
 import io.iohk.ethereum.blockchain.sync.SyncController.BlockBodiesReceived
 import io.iohk.ethereum.db.storage.AppStateStorage
 import io.iohk.ethereum.network.p2p.messages.PV62.{BlockBodies, GetBlockBodies}
+import org.spongycastle.util.encoders.Hex
 
 class SyncBlockBodiesRequestHandler(
     peer: ActorRef,
@@ -17,7 +18,8 @@ class SyncBlockBodiesRequestHandler(
 
   override def handleResponseMsg(blockBodies: BlockBodies): Unit = {
     if (blockBodies.bodies.isEmpty) {
-      syncController ! BlacklistSupport.BlacklistPeer(peer)
+      val reason = s"got empty block bodies response for known hashes: ${requestedHashes.map(h => Hex.toHexString(h.toArray[Byte]))}"
+      syncController ! BlacklistSupport.BlacklistPeer(peer, reason)
     } else {
       syncController ! BlockBodiesReceived(peer, requestedHashes, blockBodies.bodies)
     }
@@ -27,7 +29,8 @@ class SyncBlockBodiesRequestHandler(
   }
 
   override def handleTimeout(): Unit = {
-    syncController ! BlacklistSupport.BlacklistPeer(peer)
+    val reason = s"time out on block bodies response for known hashes: ${requestedHashes.map(h => Hex.toHexString(h.toArray[Byte]))}"
+    syncController ! BlacklistSupport.BlacklistPeer(peer, reason)
     syncController ! FastSync.EnqueueBlockBodies(requestedHashes)
     cleanupAndStop()
   }

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

@@ -32,14 +32,16 @@ class SyncBlockHeadersRequestHandler(
         log.info("Received {} block headers in {} ms", headers.size, timeTakenSoFar())
       }
     } else {
-      syncController ! BlacklistSupport.BlacklistPeer(peer)
+      val reason = s"got error in block headers response for requested: ${requestMsg.block}"
+      syncController ! BlacklistSupport.BlacklistPeer(peer, reason)
     }
 
     cleanupAndStop()
   }
 
   override def handleTimeout(): Unit = {
-    syncController ! BlacklistSupport.BlacklistPeer(peer)
+    val reason = s"got time out waiting for block headers response for requested: ${requestMsg.block}"
+    syncController ! BlacklistSupport.BlacklistPeer(peer, reason)
     cleanupAndStop()
   }
 

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

@@ -80,8 +80,8 @@ class SyncController(
     case Terminated(ref) if handshakedPeers.contains(ref) =>
       removePeer(ref)
 
-    case BlacklistPeer(ref) =>
-      blacklist(ref, blacklistDuration)
+    case BlacklistPeer(ref, reason) =>
+      blacklist(ref, blacklistDuration, reason)
 
     case UnblacklistPeer(ref) =>
       undoBlacklist(ref)

src/test/scala/io/iohk/ethereum/blockchain/sync/FastSyncBlockBodiesRequestHandlerSpec.scala

@@ -32,7 +32,7 @@ class FastSyncBlockBodiesRequestHandlerSpec extends FlatSpec with Matchers {
     val responseBodies = Nil
     peer.reply(PeerActor.MessageReceived(BlockBodies(responseBodies)))
 
-    parent.expectMsg(BlacklistSupport.BlacklistPeer(peer.ref))
+    parent.expectMsg(BlacklistSupport.BlacklistPeer(peer.ref, "got empty block bodies response for known hashes: List(31, 32)"))
     parent.expectMsg(SyncRequestHandler.Done)
 
     peer.expectMsg(PeerActor.Unsubscribe)
@@ -44,7 +44,7 @@ class FastSyncBlockBodiesRequestHandlerSpec extends FlatSpec with Matchers {
 
     time.advance(10.seconds)
 
-    parent.expectMsg(BlacklistSupport.BlacklistPeer(peer.ref))
+    parent.expectMsg(BlacklistSupport.BlacklistPeer(peer.ref, "time out on block bodies response for known hashes: List(31, 32)"))
     parent.expectMsg(FastSync.EnqueueBlockBodies(requestedHashes))
     parent.expectMsg(SyncRequestHandler.Done)
 

src/test/scala/io/iohk/ethereum/blockchain/sync/FastSyncBlockHeadersRequestHandlerSpec.scala

@@ -70,7 +70,7 @@ class FastSyncBlockHeadersRequestHandlerSpec extends FlatSpec with Matchers {
 
     time.advance(10.seconds)
 
-    parent.expectMsg(BlacklistSupport.BlacklistPeer(peer.ref))
+    parent.expectMsg(BlacklistSupport.BlacklistPeer(peer.ref, "got time out waiting for block headers response for requested: Left(1)"))
     parent.expectMsg(SyncRequestHandler.Done)
 
     peer.expectMsg(PeerActor.Unsubscribe)

src/test/scala/io/iohk/ethereum/blockchain/sync/FastSyncReceiptsRequestHandlerSpec.scala

@@ -34,7 +34,7 @@ class FastSyncReceiptsRequestHandlerSpec extends FlatSpec with Matchers {
 
     time.advance(10.seconds)
 
-    parent.expectMsg(BlacklistSupport.BlacklistPeer(peer.ref))
+    parent.expectMsg(BlacklistSupport.BlacklistPeer(peer.ref, "time out on receipts response for known hashes: List(31, 32)"))
     parent.expectMsg(FastSync.EnqueueReceipts(requestedHashes))
     parent.expectMsg(SyncRequestHandler.Done)