[ETCM-275] Pr comments

Move DownloaderState to separate file
Extract Actor state to separate class
Call Future.apply at call site
Improve synccontrollerspec autopilot

Author: KonradStaniec

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

@@ -0,0 +1,186 @@
+package io.iohk.ethereum.blockchain.sync
+
+import akka.util.ByteString
+import cats.data.NonEmptyList
+import io.iohk.ethereum.blockchain.sync.SyncStateScheduler.SyncResponse
+import io.iohk.ethereum.blockchain.sync.SyncStateSchedulerActor.{
+  NoUsefulDataInResponse,
+  PeerRequest,
+  ResponseProcessingResult,
+  UnrequestedResponse,
+  UsefulData
+}
+import io.iohk.ethereum.crypto.kec256
+import io.iohk.ethereum.network.{Peer, PeerId}
+import io.iohk.ethereum.network.p2p.messages.PV63.NodeData
+
+import scala.annotation.tailrec
+
+final case class DownloaderState(
+    activeRequests: Map[PeerId, NonEmptyList[ByteString]],
+    nodesToGet: Map[ByteString, Option[PeerId]]
+) {
+  lazy val nonDownloadedNodes = nodesToGet.collect {
+    case (hash, maybePeer) if maybePeer.isEmpty => hash
+  }.toSeq
+
+  def scheduleNewNodesForRetrieval(nodes: Seq[ByteString]): DownloaderState = {
+    val newNodesToGet = nodes.foldLeft(nodesToGet) { case (map, node) =>
+      if (map.contains(node)) {
+        map
+      } else {
+        map + (node -> None)
+      }
+    }
+
+    copy(nodesToGet = newNodesToGet)
+  }
+
+  private def addActiveRequest(peerRequest: PeerRequest): DownloaderState = {
+    val newNodesToget = peerRequest.nodes.foldLeft(nodesToGet) { case (map, node) =>
+      map + (node -> Some(peerRequest.peer.id))
+    }
+
+    copy(activeRequests = activeRequests + (peerRequest.peer.id -> peerRequest.nodes), nodesToGet = newNodesToget)
+  }
+
+  def handleRequestFailure(from: Peer): DownloaderState = {
+    activeRequests
+      .get(from.id)
+      .map { requestedNodes =>
+        val newNodesToGet = requestedNodes.foldLeft(nodesToGet) { case (map, node) =>
+          map + (node -> None)
+        }
+
+        copy(activeRequests = activeRequests - from.id, nodesToGet = newNodesToGet)
+      }
+      .getOrElse(this)
+  }
+
+  /**
+    * Responses from peers should be delivered in order, but can contain gaps or can be not full, so we cannot fail
+    * on first not matching response.
+    * Matched responses are returned in correct order, the hashes to be rescheduled are returned in no particular order
+    * as they will either way end up in map of hashes to be re-downloaded
+    */
+  def process(
+      requested: NonEmptyList[ByteString],
+      received: NonEmptyList[ByteString]
+  ): (List[ByteString], List[SyncResponse]) = {
+    @tailrec
+    def go(
+        remainingRequestedHashes: List[ByteString],
+        nextResponse: SyncResponse,
+        remainingResponses: List[ByteString],
+        nonReceivedRequested: List[ByteString],
+        processed: List[SyncResponse]
+    ): (List[ByteString], List[SyncResponse]) = {
+      if (remainingRequestedHashes.isEmpty) {
+        (nonReceivedRequested, processed.reverse)
+      } else {
+        val nextRequestedHash = remainingRequestedHashes.head
+        if (nextRequestedHash == nextResponse.hash) {
+          if (remainingResponses.isEmpty) {
+            val finalNonReceived = remainingRequestedHashes.tail ::: nonReceivedRequested
+            val finalProcessed = nextResponse :: processed
+            (finalNonReceived, finalProcessed.reverse)
+          } else {
+            val nexExpectedResponse = SyncResponse(kec256(remainingResponses.head), remainingResponses.head)
+            go(
+              remainingRequestedHashes.tail,
+              nexExpectedResponse,
+              remainingResponses.tail,
+              nonReceivedRequested,
+              nextResponse :: processed
+            )
+          }
+        } else {
+          go(
+            remainingRequestedHashes.tail,
+            nextResponse,
+            remainingResponses,
+            nextRequestedHash :: nonReceivedRequested,
+            processed
+          )
+        }
+      }
+    }
+
+    val firstReceivedResponse = SyncResponse(kec256(received.head), received.head)
+
+    go(requested.toList, firstReceivedResponse, received.tail, List.empty, List.empty)
+  }
+
+  def handleRequestSuccess(from: Peer, receivedMessage: NodeData): (ResponseProcessingResult, DownloaderState) = {
+    activeRequests
+      .get(from.id)
+      .map { requestedHashes =>
+        if (receivedMessage.values.isEmpty) {
+          val rescheduleRequestedHashes = requestedHashes.foldLeft(nodesToGet) { case (map, hash) =>
+            map + (hash -> None)
+          }
+          (
+            NoUsefulDataInResponse,
+            copy(activeRequests = activeRequests - from.id, nodesToGet = rescheduleRequestedHashes)
+          )
+        } else {
+          val (notReceived, received) =
+            process(requestedHashes, NonEmptyList.fromListUnsafe(receivedMessage.values.toList))
+          if (received.isEmpty) {
+            val rescheduleRequestedHashes = notReceived.foldLeft(nodesToGet) { case (map, hash) =>
+              map + (hash -> None)
+            }
+            (
+              NoUsefulDataInResponse,
+              copy(activeRequests = activeRequests - from.id, nodesToGet = rescheduleRequestedHashes)
+            )
+          } else {
+            val afterNotReceive = notReceived.foldLeft(nodesToGet) { case (map, hash) => map + (hash -> None) }
+            val afterReceived = received.foldLeft(afterNotReceive) { case (map, received) => map - received.hash }
+            (UsefulData(received), copy(activeRequests = activeRequests - from.id, nodesToGet = afterReceived))
+          }
+        }
+      }
+      .getOrElse((UnrequestedResponse, this))
+  }
+
+  def assignTasksToPeers(
+      peers: NonEmptyList[Peer],
+      newNodes: Option[Seq[ByteString]],
+      nodesPerPeerCapacity: Int
+  ): (Seq[PeerRequest], DownloaderState) = {
+    @tailrec
+    def go(
+        peersRemaining: List[Peer],
+        nodesRemaining: Seq[ByteString],
+        createdRequests: List[PeerRequest],
+        currentState: DownloaderState
+    ): (Seq[PeerRequest], DownloaderState) = {
+      if (peersRemaining.isEmpty || nodesRemaining.isEmpty) {
+        (createdRequests.reverse, currentState.scheduleNewNodesForRetrieval(nodesRemaining))
+      } else {
+        val nextPeer = peersRemaining.head
+        val (nodes, nodesAfterAssignment) = nodesRemaining.splitAt(nodesPerPeerCapacity)
+        val peerRequest = PeerRequest(nextPeer, NonEmptyList.fromListUnsafe(nodes.toList))
+        go(
+          peersRemaining.tail,
+          nodesAfterAssignment,
+          peerRequest :: createdRequests,
+          currentState.addActiveRequest(peerRequest)
+        )
+      }
+    }
+
+    val currentNodesToDeliver = newNodes.map(nodes => nonDownloadedNodes ++ nodes).getOrElse(nonDownloadedNodes)
+    if (currentNodesToDeliver.isEmpty) {
+      (Seq(), this)
+    } else {
+      go(peers.toList, currentNodesToDeliver, List.empty, this)
+    }
+  }
+
+}
+
+object DownloaderState {
+  def apply(): DownloaderState = new DownloaderState(Map.empty, Map.empty)
+}

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

@@ -0,0 +1,116 @@
+package io.iohk.ethereum.blockchain.sync
+
+import akka.actor.ActorRef
+import akka.util.ByteString
+import cats.data.NonEmptyList
+import io.iohk.ethereum.blockchain.sync.SyncStateScheduler.{ProcessingStatistics, SchedulerState}
+import io.iohk.ethereum.blockchain.sync.SyncStateSchedulerActor.{PeerRequest, RequestResult}
+import io.iohk.ethereum.network.{Peer, PeerId}
+
+import scala.collection.immutable.Queue
+
+case class SyncSchedulerActorState(
+    currentSchedulerState: SchedulerState,
+    currentDownloaderState: DownloaderState,
+    currentStats: ProcessingStatistics,
+    targetBlock: BigInt,
+    syncInitiator: ActorRef,
+    nodesToProcess: Queue[RequestResult],
+    processing: Boolean,
+    restartRequested: Option[ActorRef]
+) {
+  def hasRemainingPendingRequests: Boolean = currentSchedulerState.numberOfPendingRequests > 0
+  def isProcessing: Boolean = processing
+  def restartHasBeenRequested: Boolean = restartRequested.isDefined
+  def withNewRequestResult(requestResult: RequestResult): SyncSchedulerActorState =
+    copy(nodesToProcess = nodesToProcess.enqueue(requestResult))
+
+  def withNewProcessingResults(
+      newSchedulerState: SchedulerState,
+      newDownloaderState: DownloaderState,
+      newStats: ProcessingStatistics
+  ): SyncSchedulerActorState = {
+    copy(
+      currentSchedulerState = newSchedulerState,
+      currentDownloaderState = newDownloaderState,
+      currentStats = newStats
+    )
+  }
+
+  def withNewDownloaderState(newDownloaderState: DownloaderState): SyncSchedulerActorState = {
+    copy(currentDownloaderState = newDownloaderState)
+  }
+
+  def withRestartRequested(restartRequester: ActorRef): SyncSchedulerActorState = {
+    copy(restartRequested = Some(restartRequester))
+  }
+
+  def initProcessing: SyncSchedulerActorState = {
+    copy(processing = true)
+  }
+
+  def finishProcessing: SyncSchedulerActorState = {
+    copy(processing = false)
+  }
+
+  def assignTasksToPeers(
+      freePeers: NonEmptyList[Peer],
+      nodesPerPeer: Int
+  ): (Seq[PeerRequest], SyncSchedulerActorState) = {
+    val retryQueue = currentDownloaderState.nonDownloadedNodes
+    val maxNewNodes = ((freePeers.size * nodesPerPeer) - retryQueue.size).max(0)
+    val (newNodes, newState) = currentSchedulerState.getMissingHashes(maxNewNodes)
+    val (requests, newDownloaderState) =
+      currentDownloaderState.assignTasksToPeers(
+        NonEmptyList.fromListUnsafe(freePeers.toList),
+        Some(newNodes),
+        nodesPerPeer
+      )
+    (requests, copy(currentSchedulerState = newState, currentDownloaderState = newDownloaderState))
+  }
+
+  def getRequestToProcess: Option[(RequestResult, SyncSchedulerActorState)] = {
+    nodesToProcess.dequeueOption.map { case (result, restOfResults) =>
+      (result, copy(nodesToProcess = restOfResults))
+    }
+  }
+
+  def numberOfRemainingRequests: Int = nodesToProcess.size
+
+  def memBatch: Map[ByteString, (ByteString, SyncStateScheduler.RequestType)] = currentSchedulerState.memBatch
+
+  def activePeerRequests: Map[PeerId, NonEmptyList[ByteString]] = currentDownloaderState.activeRequests
+
+  override def toString: String = {
+    s""" Status of mpt state sync:
+       | Number of Pending requests: ${currentSchedulerState.numberOfPendingRequests},
+       | Number of Missing hashes waiting to be retrieved: ${currentSchedulerState.queue.size()},
+       | Number of Requests waiting for processing: ${nodesToProcess.size},
+       | Number of Mpt nodes saved to database: ${currentStats.saved},
+       | Number of duplicated hashes: ${currentStats.duplicatedHashes},
+       | Number of not requested hashes: ${currentStats.notRequestedHashes},
+       | Number of active peer requests: ${currentDownloaderState.activeRequests.size}
+                        """.stripMargin
+  }
+}
+
+object SyncSchedulerActorState {
+  def initial(
+      initialSchedulerState: SchedulerState,
+      initialStats: ProcessingStatistics,
+      targetBlock: BigInt,
+      syncInitiator: ActorRef
+  ): SyncSchedulerActorState = {
+    SyncSchedulerActorState(
+      initialSchedulerState,
+      DownloaderState(),
+      initialStats,
+      targetBlock,
+      syncInitiator,
+      Queue(),
+      processing = false,
+      restartRequested = None
+    )
+
+  }
+}