refactor FetcherService to use buildBlocks, implement placeBlockInPeerStream via SourceQueue

Author: Jaap van der Plas

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

@@ -25,9 +25,17 @@ import io.iohk.ethereum.network.PeerId
 import io.iohk.ethereum.network.p2p.messages.ETH62.BlockBodies
 import io.iohk.ethereum.network.p2p.messages.ETH62.BlockHeaders
 import io.iohk.ethereum.utils.Config.SyncConfig
+import akka.stream.scaladsl.SourceQueue
+import akka.stream.QueueOfferResult.Enqueued
 
 //not used atm, a part of the future ExecutionSync
-class FetcherService(validator: BlockValidator, blockchainReader: BlockchainReader, syncConfig: SyncConfig) {
+class FetcherService(
+    blockchainReader: BlockchainReader,
+    syncConfig: SyncConfig,
+    sourceQueue: SourceQueue[Block]
+) {
+
+  import FetcherService._
 
   val batchSize = syncConfig.blockHeadersPerRequest
 
@@ -40,7 +48,11 @@ class FetcherService(validator: BlockValidator, blockchainReader: BlockchainRead
 
 //TODO: add private def requestStateNode(hash: ByteString): Task[Either[RequestFailed, Seq[ByteString]]] = ???
 
-  private def placeBlockInPeerStream(block: Block, peer: Peer): Peer = ???
+  private def placeBlockInPeerStream(peer: Peer, block: Block): Task[Either[RequestFailed, Unit]] =
+    Task.deferFuture(sourceQueue.offer(block)).map {
+      case Enqueued => Right(())
+      case result   => Left(RequestFailed(peer, result.toString()))
+    }
 
   def fetchBlocksUntil(
       peer: Peer,
@@ -73,30 +85,10 @@ class FetcherService(validator: BlockValidator, blockchainReader: BlockchainRead
     for {
       headers <- EitherT(requestHeaders(block, amount))
       bodies <- EitherT(requestBodies(headers.headers.map(_.hash)))
-      blocks <- EitherT.fromOption[Task](
-        bodiesAreOrderedSubsetOfRequested(headers.headers, bodies.bodies),
-        RequestFailed(peer, "Unmatching bodies")
-      )
-      _ = blocks.foreach(placeBlockInPeerStream(_, peer))
+      blocks = buildBlocks(headers.headers, bodies.bodies)
+      _ <- EitherT.cond[Task](blocks.length == headers.headers.length, (), RequestFailed(peer, "Unmatching bodies"))
+      _ <- blocks.traverse(block => EitherT(placeBlockInPeerStream(peer, block)))
     } yield peer
-
-  // Checks that the received block bodies are an ordered subset of the ones requested
-  @tailrec
-  private def bodiesAreOrderedSubsetOfRequested(
-      requestedHeaders: Seq[BlockHeader],
-      respondedBodies: Seq[BlockBody],
-      matchedBlocks: Seq[Block] = Nil
-  ): Option[Seq[Block]] =
-    (requestedHeaders, respondedBodies) match {
-      case (Seq(), _ +: _) => None
-      case (_, Seq())      => Some(matchedBlocks)
-      case (header +: remainingHeaders, body +: remainingBodies) =>
-        if (validator.validateHeaderAndBody(header, body).isRight)
-          bodiesAreOrderedSubsetOfRequested(remainingHeaders, remainingBodies, matchedBlocks :+ Block(header, body))
-        else
-          bodiesAreOrderedSubsetOfRequested(remainingHeaders, respondedBodies, matchedBlocks)
-    }
-
 }
 
 object FetcherService {

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

@@ -41,6 +41,7 @@ import io.iohk.ethereum.network.p2p.messages.ETH62.BlockBodies
 import io.iohk.ethereum.network.p2p.messages.ETH62.BlockHeaders
 import io.iohk.ethereum.utils.ByteStringUtils
 import io.iohk.ethereum.utils.Hex
+import akka.stream.scaladsl.SourceQueue
 
 class FetcherServiceSpec
     extends TestKit(ActorSystem("FetcherServiceSpec_System"))
@@ -60,8 +61,8 @@ class FetcherServiceSpec
   "FetcherService" should "return RequestFailed when asking for a not existing blocks" in {
     val blockchainReader: BlockchainReader = mock[BlockchainReader]
     (blockchainReader.getBlockHeaderByHash _).expects(*).returning(None).anyNumberOfTimes()
-    val blockValidator: BlockValidator = mock[BlockValidator]
-    val fetcherService = new FetcherService(blockValidator, blockchainReader, syncConfig)
+    val sourceQueue = mock[SourceQueue[Block]]
+    val fetcherService = new FetcherService(blockchainReader, syncConfig, sourceQueue)
     val peerProbe: TestProbe = TestProbe()
     val peer = Peer(PeerId("peerId"), new InetSocketAddress(9191), peerProbe.ref, true)
     val result = fetcherService.fetchBlocksUntil(peer, Right(ByteString("byteString")), Right(ByteString("byteString")))
@@ -72,40 +73,13 @@ class FetcherServiceSpec
   it should "return an error when request to fetch headers in fetchBlocks fails" ignore {
     val blockchainReader: BlockchainReader = mock[BlockchainReader]
     (blockchainReader.getHashByBlockNumber _).expects(*, *).returning(Some(genesisHash))
-    val blockValidator: BlockValidator = mock[BlockValidator]
-    val fetcherService = new FetcherService(blockValidator, blockchainReader, syncConfig)
+    val sourceQueue = mock[SourceQueue[Block]]
+    val fetcherService = new FetcherService(blockchainReader, syncConfig, sourceQueue)
     val fetchBlocks = PrivateMethod[EitherT[Task, RequestFailed, Peer]]('fetchBlocks)
     val eitherPeerOrError = fetcherService.invokePrivate(fetchBlocks())
     assert(eitherPeerOrError === RequestFailed)
   }
 
-  it should "fail to verify that bodies are a subset of headers if they don't match" in {
-    val blockchainReader: BlockchainReader = mock[BlockchainReader]
-    // Here we are forcing the mismatch between request headers and received bodies
-    val blockValidator = new MockValidatorsFailingOnBlockBodies
-    val fetcherService = new FetcherService(blockValidator.blockValidator, blockchainReader, syncConfig)
-    val bodiesAreOrderedSubsetOfRequested = PrivateMethod[Option[Seq[Block]]]('bodiesAreOrderedSubsetOfRequested)
-    val blocks = Seq(Block(ValidBlock.header.copy(number = 97), ValidBlock.body))
-    val headers: Seq[BlockHeader] = blocks.map(_.header)
-    val bodies: Seq[BlockBody] = blocks.map(_.body)
-    val verifiedBlocks: Option[Seq[Block]] =
-      fetcherService.invokePrivate(bodiesAreOrderedSubsetOfRequested(headers, bodies, Nil))
-    verifiedBlocks shouldBe None
-  }
-
-  it should "verify that bodies are a subset of headers" in {
-    val blockchainReader: BlockchainReader = mock[BlockchainReader]
-    val blockValidator = new MockValidatorsAlwaysSucceed
-    val fetcherService = new FetcherService(blockValidator.blockValidator, blockchainReader, syncConfig)
-    val bodiesAreOrderedSubsetOfRequested = PrivateMethod[Option[Seq[Block]]]('bodiesAreOrderedSubsetOfRequested)
-    val blocks = Seq(Block(ValidBlock.header.copy(number = 97), ValidBlock.body))
-    val headers: Seq[BlockHeader] = blocks.map(_.header)
-    val bodies: Seq[BlockBody] = blocks.map(_.body)
-    val verifiedBlocks: Option[Seq[Block]] =
-      fetcherService.invokePrivate(bodiesAreOrderedSubsetOfRequested(headers, bodies, Nil))
-    verifiedBlocks shouldBe Some(blocks)
-  }
-
   val emptyTransactionsRoot: ByteString =
     ByteStringUtils.string2hash("56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421")
   val emptyOmmersHash: ByteString =
@@ -147,6 +121,11 @@ class FetcherServiceSpec
     (FetcherService.BlockIdentifier(body1) should not).equal(FetcherService.BlockIdentifier(header2))
   }
 
+  "FetcherService.buildBlocks" should "return matching headers and bodies only" in {
+    val result = FetcherService.buildBlocks(Seq(header1, header2), Seq(body2, body3, body1))
+    result shouldEqual Seq(Block(header1, body1), Block(header2, body2))
+  }
+
   "FetcherService.fetchBlocksForHeaders" should "combine matching headers and bodies" in {
     val messages = Seq(
       MessageFromPeer(BlockHeaders(Seq(header1, header2)), peerId),