[FIX] Handle timeouts on BlockFetcher

Author: Nicolas Tallar

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcher.scala

@@ -9,10 +9,7 @@ import cats.instances.future._
 import cats.instances.option._
 import cats.syntax.either._
 import io.iohk.ethereum.blockchain.sync.PeersClient._
-import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState.{
-  AwaitingBodiesToBeIgnored,
-  AwaitingHeadersToBeIgnored
-}
+import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState.{AwaitingBodiesToBeIgnored, AwaitingHeadersToBeIgnored}
 import io.iohk.ethereum.blockchain.sync.regular.BlockImporter.{ImportNewBlock, NotOnTop, OnTop}
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.ProgressProtocol
 import io.iohk.ethereum.crypto.kec256
@@ -44,7 +41,7 @@ class BlockFetcher(
   import BlockFetcher._
 
   implicit val ec: ExecutionContext = context.dispatcher
-  implicit val timeout: Timeout = syncConfig.peerResponseTimeout + 1.second // some margin for actor communication
+  implicit val timeout: Timeout = syncConfig.peerResponseTimeout + 2.second // some margin for actor communication
 
   override def receive: Receive = idle()
 
@@ -130,7 +127,7 @@ class BlockFetcher(
 
       fetchBlocks(newState)
     case RetryHeadersRequest if state.isFetchingHeaders =>
-      log.debug("Time-out occurred while waiting for headers")
+      log.debug("Something failed on a headers request, cancelling the request and re-fetching")
 
       val newState = state.withHeaderFetchReceived
       fetchBlocks(newState)
@@ -149,7 +146,7 @@ class BlockFetcher(
 
       fetchBlocks(newState)
     case RetryBodiesRequest if state.isFetchingBodies =>
-      log.debug("Time-out occurred while waiting for bodies")
+      log.debug("Something failed on a bodies request, cancelling the request and re-fetching")
 
       val newState = state.withBodiesFetchReceived
       fetchBlocks(newState)
@@ -316,6 +313,10 @@ class BlockFetcher(
     (peersClient ? request)
       .tap(blacklistPeerOnFailedRequest)
       .flatMap(failureTo(responseFallback))
+      .recover { case error =>
+        log.error(error, "Unexpected error on a request")
+        responseFallback
+      }
 
   private def blacklistPeerOnFailedRequest(msg: Any): Unit = msg match {
     case RequestFailed(peer, reason) => peersClient ! BlacklistPeer(peer.id, reason)

src/test/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcherSpec.scala

@@ -195,6 +195,24 @@ class BlockFetcherSpec extends TestKit(ActorSystem("BlockFetcherSpec_System")) w
         assert(HeadersSeq.areChain(headers))
       }
     }
+
+    "should properly handle a request timeout" in new TestSetup {
+      override lazy val syncConfig = defaultSyncConfig.copy(
+        // Small timeout on ask pattern for testing it here
+        peerResponseTimeout = 3.seconds
+      )
+
+      startFetcher()
+
+      val firstGetBlockHeadersRequest =
+        GetBlockHeaders(Left(1), syncConfig.blockHeadersPerRequest, skip = 0, reverse = false)
+      peersClient.expectMsgPF() { case PeersClient.Request(msg, _, _) if msg == firstGetBlockHeadersRequest => () }
+
+      // Request should timeout without any response from the peer
+      Thread.sleep((2 * syncConfig.peerResponseTimeout).toMillis)
+
+      peersClient.expectMsgPF() { case PeersClient.Request(msg, _, _) if msg == firstGetBlockHeadersRequest => () }
+    }
   }
 
   trait TestSetup extends TestSyncConfig {