[FIX] Handle timeouts on BlockFetcher (#805)

Author: Nicolás Tallar

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

@@ -44,7 +44,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 +130,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 +149,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)
@@ -315,19 +315,23 @@ class BlockFetcher(
   private def makeRequest(request: Request[_], responseFallback: FetchMsg): Future[Any] =
     (peersClient ? request)
       .tap(blacklistPeerOnFailedRequest)
-      .flatMap(failureTo(responseFallback))
+      .flatMap(handleRequestResult(responseFallback))
+      .recover { case error =>
+        log.error(error, "Unexpected error while doing a request")
+        responseFallback
+      }
 
   private def blacklistPeerOnFailedRequest(msg: Any): Unit = msg match {
     case RequestFailed(peer, reason) => peersClient ! BlacklistPeer(peer.id, reason)
     case _ => ()
   }
 
-  private def failureTo(fallback: FetchMsg)(msg: Any): Future[Any] = msg match {
+  private def handleRequestResult(fallback: FetchMsg)(msg: Any): Future[Any] = msg match {
     case failed: RequestFailed =>
-      log.debug("Failed request {}", failed)
+      log.debug("Request failed due to {}", failed)
       Future.successful(fallback)
     case Failure(cause) =>
-      log.debug("Failed request due to {}", cause)
+      log.error(cause, "Unexpected error on the request result")
       Future.successful(fallback)
     case NoSuitablePeer =>
       Future.successful(fallback).delayedBy(syncConfig.syncRetryInterval)

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 = 1.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((syncConfig.peerResponseTimeout + 2.seconds).toMillis)
+
+      peersClient.expectMsgPF() { case PeersClient.Request(msg, _, _) if msg == firstGetBlockHeadersRequest => () }
+    }
   }
 
   trait TestSetup extends TestSyncConfig {