[Etcm 521] Fast sync integration tests (#944)

* [ETCM-521] Add integration tests for fast sync
- Apply fix for branch resolving not being started

* [ETCM-521] Fix failing test, cleanup

* [ETCM-521] Fix failing test

* [ETCM-521] Address review comments

* Add comments, rename helper methods

Author: robinraju

src/it/scala/io/iohk/ethereum/sync/FastSyncItSpec.scala

@@ -2,8 +2,10 @@ package io.iohk.ethereum.sync
 
 import akka.util.ByteString
 import io.iohk.ethereum.FlatSpecBase
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason.BlacklistReasonType
 import io.iohk.ethereum.domain._
 import io.iohk.ethereum.ledger.InMemoryWorldStateProxy
+import io.iohk.ethereum.network.PeerId
 import io.iohk.ethereum.sync.FastSyncItSpec._
 import io.iohk.ethereum.sync.util.FastSyncItSpecUtils.FakePeer
 import io.iohk.ethereum.sync.util.SyncCommonItSpec._
@@ -180,6 +182,65 @@ class FastSyncItSpec extends FlatSpecBase with Matchers with BeforeAndAfterAll {
     }
   }
 
+  it should "switch to regular sync once `safeDownloadTarget` is reached" in customTestCaseResourceM(
+    FakePeer.start3FakePeersRes()
+  ) { case (peer1, peer2, peer3) =>
+    for {
+      _ <- peer2.importBlocksUntil(1200)(IdentityUpdate)
+      _ <- peer3.importBlocksUntil(1200)(IdentityUpdate)
+      _ <- peer1.connectToPeers(Set(peer2.node, peer3.node))
+      _ <- peer1.startFastSync().delayExecution(50.milliseconds)
+      _ <- peer1.waitForFastSyncFinish()
+    } yield {
+      assert(peer1.bl.getBestBlockNumber() == peer3.bl.getBestBlockNumber() - peer3.testSyncConfig.pivotBlockOffset)
+    }
+  }
+
+  it should "blacklist peer on Invalid batch last header number" in customTestCaseResourceM(
+    FakePeer.start3FakePeersRes()
+  ) { case (peer1, peer2, peer3) =>
+    for {
+      _ <- peer2.importBlocksUntil(1000)(IdentityUpdate)
+      _ <- peer3.importInvalidBlockNumbers(201, 1200)(IdentityUpdate)
+
+      _ <- peer1.connectToPeers(Set(peer2.node, peer3.node))
+      _ <- peer1.startFastSync().delayExecution(50.milliseconds)
+      _ <- peer1.waitForFastSyncFinish()
+    } yield {
+      // Peer3 is blacklisted
+      val blacklistedPeer = PeerId(s"${peer3.node.addr.getHostAddress}:${peer3.node.tcpPort}")
+      val blacklistReason = peer1.blacklist.cache.getIfPresent(blacklistedPeer)
+
+      assert(peer1.blacklist.isBlacklisted(blacklistedPeer))
+      assert(blacklistReason.get == BlacklistReasonType.BlockHeaderValidationFailedType)
+    }
+  }
+
+  it should "sync blockchain when peer responds with invalid batch last header hash" in customTestCaseResourceM(
+    FakePeer.start4FakePeersRes()
+  ) { case (peer1, peer2, peer3, peer4) =>
+    for {
+      _ <- peer1.importBlocksUntil(400)(IdentityUpdate)
+      _ <- peer2.importBlocksUntil(1000)(IdentityUpdate)
+
+      _ <- peer3.importInvalidBlocks(600, 800)(IdentityUpdate)
+      _ <- peer3.importBlocksUntil(1200)(updateStateAtBlock(1000))
+
+      _ <- peer4.importBlocksUntil(1100)(IdentityUpdate)
+
+      _ <- peer1.connectToPeers(Set(peer2.node, peer3.node, peer4.node))
+      _ <- peer1.startFastSync().delayExecution(50.milliseconds)
+      _ <- peer2.importBlocksUntil(1200)(IdentityUpdate).startAndForget
+      _ <- peer1.waitForFastSyncFinish()
+    } yield {
+      // Peer3 is blacklisted
+      val blacklistedPeer = PeerId(s"${peer3.node.addr.getHostAddress}:${peer3.node.tcpPort}")
+      val blacklistReason = peer1.blacklist.cache.getIfPresent(blacklistedPeer)
+
+      assert(peer1.blacklist.isBlacklisted(blacklistedPeer))
+      assert(blacklistReason.get == BlacklistReasonType.BlockHeaderValidationFailedType)
+    }
+  }
 }
 
 object FastSyncItSpec {

src/it/scala/io/iohk/ethereum/sync/util/CommonFakePeer.scala

@@ -223,7 +223,9 @@ abstract class CommonFakePeer(peerName: String, fakePeerCustomConfig: FakePeerCu
     nodesPerRequest = 200,
     maxTargetDifference = 1,
     syncRetryInterval = 50.milliseconds,
-    blacklistDuration = 100.seconds
+    blacklistDuration = 100.seconds,
+    fastSyncMaxBatchRetries = 2,
+    fastSyncBlockValidationN = 200
   )
 
   lazy val broadcaster = new BlockBroadcast(etcPeerManager)
@@ -298,21 +300,90 @@ abstract class CommonFakePeer(peerName: String, fakePeerCustomConfig: FakePeerCu
     (newBlock, newWeight, parentWorld)
   }
 
+  private def generateInvalidBlock(
+      currentBestBlock: Block
+  )(updateWorldForBlock: (BigInt, InMemoryWorldStateProxy) => InMemoryWorldStateProxy): Task[Unit] = {
+    Task {
+      val currentWorld = getMptForBlock(currentBestBlock)
+
+      val newBlockNumber = currentBestBlock.header.number + 1
+      val newWorld = updateWorldForBlock(newBlockNumber, currentWorld)
+
+      // The child block is made invalid by not properly updating its parent hash.
+      val childBlock =
+        currentBestBlock.copy(header =
+          currentBestBlock.header.copy(
+            number = newBlockNumber,
+            stateRoot = newWorld.stateRootHash
+          )
+        )
+      val newWeight = ChainWeight.totalDifficultyOnly(1)
+
+      broadcastBlock(childBlock, newWeight)
+      bl.save(childBlock, Seq(), newWeight, saveAsBestBlock = true)
+    }
+  }
+
+  private def generateValidBlock(
+      currentBestBlock: Block
+  )(updateWorldForBlock: (BigInt, InMemoryWorldStateProxy) => InMemoryWorldStateProxy): Task[Unit] = {
+    Task {
+      val currentWeight = bl.getChainWeightByHash(currentBestBlock.hash).get
+      val currentWorld = getMptForBlock(currentBestBlock)
+      val (newBlock, newWeight, _) =
+        createChildBlock(currentBestBlock, currentWeight, currentWorld)(updateWorldForBlock)
+      bl.save(newBlock, Seq(), newWeight, saveAsBestBlock = true)
+      broadcastBlock(newBlock, newWeight)
+    }
+  }
+
   def importBlocksUntil(
       n: BigInt
   )(updateWorldForBlock: (BigInt, InMemoryWorldStateProxy) => InMemoryWorldStateProxy): Task[Unit] = {
     Task(bl.getBestBlock()).flatMap { block =>
       if (block.get.number >= n) {
         Task(())
       } else {
-        Task {
-          val currentWeight = bl.getChainWeightByHash(block.get.hash).get
-          val currentWolrd = getMptForBlock(block.get)
-          val (newBlock, newWeight, _) = createChildBlock(block.get, currentWeight, currentWolrd)(updateWorldForBlock)
-          bl.save(newBlock, Seq(), newWeight, saveAsBestBlock = true)
-          broadcastBlock(newBlock, newWeight)
-        }.flatMap(_ => importBlocksUntil(n)(updateWorldForBlock))
+        generateValidBlock(block.get)(updateWorldForBlock).flatMap(_ => importBlocksUntil(n)(updateWorldForBlock))
+      }
+    }
+  }
+
+  def importInvalidBlocks(
+      from: BigInt,
+      to: BigInt
+  )(updateWorldForBlock: (BigInt, InMemoryWorldStateProxy) => InMemoryWorldStateProxy): Task[Unit] = {
+    Task(bl.getBestBlock()).flatMap { block =>
+      if (block.get.number >= to) {
+        Task(())
+      } else if (block.get.number >= from) {
+        generateInvalidBlock(block.get)(updateWorldForBlock).flatMap(_ =>
+          importInvalidBlocks(from, to)(updateWorldForBlock)
+        )
+      } else {
+        generateValidBlock(block.get)(updateWorldForBlock).flatMap(_ =>
+          importInvalidBlocks(from, to)(updateWorldForBlock)
+        )
       }
+
+    }
+  }
+
+  def importInvalidBlockNumbers(
+      from: BigInt,
+      to: BigInt
+  )(updateWorldForBlock: (BigInt, InMemoryWorldStateProxy) => InMemoryWorldStateProxy): Task[Unit] = {
+    Task(bl.getBestBlock()).flatMap { block =>
+      if (block.get.number >= to) {
+        Task(())
+      } else if (block.get.number >= from) {
+        generateInvalidBlock(block.get)(updateWorldForBlock).flatMap(_ =>
+          importInvalidBlockNumbers(from, to)(updateWorldForBlock)
+        )
+      } else {
+        importBlocksUntil(from)(updateWorldForBlock)
+      }
+
     }
   }
 

src/main/scala/io/iohk/ethereum/blockchain/sync/fast/FastSyncBranchResolverActor.scala

@@ -79,7 +79,7 @@ class FastSyncBranchResolverActor(
     handlePeerListMessages orElse {
       case ResponseReceived(peer, BlockHeaders(headers), durationMs) if peer == searchState.masterPeer =>
         context.unwatch(requestHandler)
-        headers match {
+        headers.toList match {
           case childHeader :: Nil if childHeader.number == blockHeaderNumberToSearch =>
             log.debug(ReceivedBlockHeaderLog, blockHeaderNumberToSearch, peer.id, durationMs)
             handleBinarySearchBlockHeaderResponse(searchState, childHeader)