[ETCM-280] use peer capabilities (protocol version) when sending checkpoint-related messages

Author: rtkaczyk

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

@@ -6,6 +6,7 @@ import java.util.concurrent.atomic.AtomicReference
 import akka.actor.{ActorRef, ActorSystem}
 import akka.testkit.TestProbe
 import akka.util.{ByteString, Timeout}
+import io.iohk.ethereum.blockchain.sync.BlockBroadcast.BlockToBroadcast
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcasterActor
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcasterActor.BroadcastBlock
 import io.iohk.ethereum.blockchain.sync.{BlockBroadcast, BlockchainHostActor, TestSyncConfig}
@@ -18,21 +19,13 @@ import io.iohk.ethereum.ledger.InMemoryWorldStateProxy
 import io.iohk.ethereum.mpt.MerklePatriciaTrie
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.PeerManagerActor.{FastSyncHostConfiguration, PeerConfiguration}
-import io.iohk.ethereum.network.discovery.{DiscoveryConfig, Node}
 import io.iohk.ethereum.network.discovery.PeerDiscoveryManager.{DiscoveredNodesInfo, DiscoveryNodeInfo}
+import io.iohk.ethereum.network.discovery.{DiscoveryConfig, Node}
 import io.iohk.ethereum.network.handshaker.{EtcHandshaker, EtcHandshakerConfiguration, Handshaker}
 import io.iohk.ethereum.network.p2p.EthereumMessageDecoder
-import io.iohk.ethereum.network.p2p.messages.CommonMessages.NewBlock
 import io.iohk.ethereum.network.rlpx.AuthHandshaker
 import io.iohk.ethereum.network.rlpx.RLPxConnectionHandler.RLPxConfiguration
-import io.iohk.ethereum.network.{
-  EtcPeerManagerActor,
-  ForkResolver,
-  KnownNodesManager,
-  PeerEventBusActor,
-  PeerManagerActor,
-  ServerActor
-}
+import io.iohk.ethereum.network.{EtcPeerManagerActor, ForkResolver, KnownNodesManager, PeerEventBusActor, PeerManagerActor, ServerActor}
 import io.iohk.ethereum.nodebuilder.{PruningConfigBuilder, SecureRandomBuilder}
 import io.iohk.ethereum.sync.util.SyncCommonItSpec._
 import io.iohk.ethereum.sync.util.SyncCommonItSpecUtils._
@@ -159,7 +152,7 @@ abstract class CommonFakePeer(peerName: String, fakePeerCustomConfig: FakePeerCu
       override val peerConfiguration: PeerConfiguration = peerConf
       override val blockchain: Blockchain = bl
       override val appStateStorage: AppStateStorage = storagesInstance.storages.appStateStorage
-      override val blockchainConfig = CommonFakePeer.this.blockchainConfig // FIXME: remove in ETCM-280
+      override val protocolVersion: Int = Config.Network.protocolVersion
     }
 
   lazy val handshaker: Handshaker[PeerInfo] = EtcHandshaker(handshakerConfiguration)
@@ -230,7 +223,7 @@ abstract class CommonFakePeer(peerName: String, fakePeerCustomConfig: FakePeerCu
   }
 
   private def broadcastBlock(block: Block, weight: ChainWeight) = {
-    broadcasterActor ! BroadcastBlock(NewBlock(block, weight))
+    broadcasterActor ! BroadcastBlock(BlockToBroadcast(block, weight))
   }
 
   def getCurrentState(): BlockchainState = {

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

@@ -4,15 +4,15 @@ import akka.actor.ActorRef
 import akka.util.ByteString
 import cats.effect.Resource
 import io.iohk.ethereum.Mocks.MockValidatorsAlwaysSucceed
-import io.iohk.ethereum.blockchain.sync.{PeersClient, SyncProtocol}
+import io.iohk.ethereum.blockchain.sync.BlockBroadcast.BlockToBroadcast
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcasterActor.BroadcastBlock
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync
+import io.iohk.ethereum.blockchain.sync.{PeersClient, SyncProtocol}
 import io.iohk.ethereum.consensus.blocks.CheckpointBlockGenerator
 import io.iohk.ethereum.consensus.ethash.{EthashConfig, EthashConsensus}
 import io.iohk.ethereum.consensus.{ConsensusConfig, FullConsensusConfig, ethash}
 import io.iohk.ethereum.domain._
 import io.iohk.ethereum.ledger._
-import io.iohk.ethereum.network.p2p.messages.CommonMessages.NewBlock
 import io.iohk.ethereum.nodebuilder.VmSetup
 import io.iohk.ethereum.ommers.OmmersPool
 import io.iohk.ethereum.sync.util.SyncCommonItSpecUtils.FakePeerCustomConfig.defaultConfig
@@ -70,7 +70,6 @@ object RegularSyncItSpecUtils {
         peerEventBus,
         ledger,
         bl,
-        blockchainConfig, // FIXME: remove in ETCM-280
         testSyncConfig,
         ommersPool,
         pendingTransactionsManager,
@@ -140,7 +139,7 @@ object RegularSyncItSpecUtils {
     }
 
     private def broadcastBlock(block: Block, weight: ChainWeight) = {
-      broadcasterActor ! BroadcastBlock(NewBlock(block, weight))
+      broadcasterActor ! BroadcastBlock(BlockToBroadcast(block, weight))
     }
 
     private def createChildBlock(

src/it/scala/io/iohk/ethereum/txExecTest/util/DumpChainApp.scala

@@ -1,34 +1,34 @@
 package io.iohk.ethereum.txExecTest.util
 
+import java.util.concurrent.atomic.AtomicReference
+
 import akka.actor.ActorSystem
 import akka.util.ByteString
 import com.typesafe.config.ConfigFactory
 import io.iohk.ethereum.db.components.Storages.PruningModeComponent
 import io.iohk.ethereum.db.components.{RocksDbDataSourceComponent, Storages}
-import io.iohk.ethereum.db.storage.{AppStateStorage, StateStorage}
+import io.iohk.ethereum.db.dataSource.{DataSourceBatchUpdate, RocksDbDataSource}
 import io.iohk.ethereum.db.storage.NodeStorage.{NodeEncoded, NodeHash}
 import io.iohk.ethereum.db.storage.TransactionMappingStorage.TransactionLocation
 import io.iohk.ethereum.db.storage.pruning.{ArchivePruning, PruningMode}
+import io.iohk.ethereum.db.storage.{AppStateStorage, StateStorage}
+import io.iohk.ethereum.domain.BlockHeader.HeaderExtraFields.HefEmpty
 import io.iohk.ethereum.domain.{Blockchain, UInt256, _}
 import io.iohk.ethereum.ledger.{InMemoryWorldStateProxy, InMemoryWorldStateProxyStorage}
 import io.iohk.ethereum.mpt.MptNode
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.PeerManagerActor.PeerConfiguration
+import io.iohk.ethereum.network.discovery.DiscoveryConfig
 import io.iohk.ethereum.network.handshaker.{EtcHandshaker, EtcHandshakerConfiguration, Handshaker}
 import io.iohk.ethereum.network.p2p.EthereumMessageDecoder
 import io.iohk.ethereum.network.rlpx.RLPxConnectionHandler.RLPxConfiguration
 import io.iohk.ethereum.network.{ForkResolver, PeerEventBusActor, PeerManagerActor}
 import io.iohk.ethereum.nodebuilder.{AuthHandshakerBuilder, NodeKeyBuilder, SecureRandomBuilder}
-import io.iohk.ethereum.utils.{BlockchainConfig, Config, NodeStatus, ServerStatus}
-import java.util.concurrent.atomic.AtomicReference
-
-import io.iohk.ethereum.db.dataSource.{DataSourceBatchUpdate, RocksDbDataSource}
+import io.iohk.ethereum.utils.{Config, NodeStatus, ServerStatus}
+import monix.reactive.Observable
 import org.bouncycastle.util.encoders.Hex
 
 import scala.concurrent.duration._
-import io.iohk.ethereum.domain.BlockHeader.HeaderExtraFields.HefEmpty
-import io.iohk.ethereum.network.discovery.DiscoveryConfig
-import monix.reactive.Observable
 
 object DumpChainApp extends App with NodeKeyBuilder with SecureRandomBuilder with AuthHandshakerBuilder {
   val conf = ConfigFactory.load("txExecTest/chainDump.conf")
@@ -82,8 +82,8 @@ object DumpChainApp extends App with NodeKeyBuilder with SecureRandomBuilder wit
       override val nodeStatusHolder: AtomicReference[NodeStatus] = DumpChainApp.nodeStatusHolder
       override val peerConfiguration: PeerConfiguration = peerConfig
       override val blockchain: Blockchain = DumpChainApp.blockchain
-      override val blockchainConfig: BlockchainConfig = DumpChainApp.blockchainConfig
       override val appStateStorage: AppStateStorage = storagesInstance.storages.appStateStorage
+      override val protocolVersion: Int = Config.Network.protocolVersion
     }
 
   lazy val handshaker: Handshaker[PeerInfo] = EtcHandshaker(handshakerConfiguration)

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

@@ -1,11 +1,13 @@
 package io.iohk.ethereum.blockchain.sync
 
 import akka.actor.ActorRef
-import io.iohk.ethereum.network.{EtcPeerManagerActor, Peer}
+import io.iohk.ethereum.blockchain.sync.BlockBroadcast.BlockToBroadcast
+import io.iohk.ethereum.domain.{Block, ChainWeight}
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
-import io.iohk.ethereum.network.p2p.messages.CommonMessages.NewBlock
-import io.iohk.ethereum.network.p2p.messages.PV62
+import io.iohk.ethereum.network.p2p.MessageSerializable
 import io.iohk.ethereum.network.p2p.messages.PV62.BlockHash
+import io.iohk.ethereum.network.p2p.messages.{CommonMessages, PV62, PV64, Versions}
+import io.iohk.ethereum.network.{EtcPeerManagerActor, Peer}
 import io.iohk.ethereum.utils.Config.SyncConfig
 
 import scala.util.Random
@@ -21,29 +23,31 @@ class BlockBroadcast(val etcPeerManager: ActorRef, syncConfig: SyncConfig) {
     * @param newBlock, block to broadcast
     * @param handshakedPeers, to which the blocks will be broadcasted to
     */
-  def broadcastBlock(newBlock: NewBlock, handshakedPeers: Map[Peer, PeerInfo]): Unit = {
-    val peersWithoutBlock = handshakedPeers.collect {
-      case (peer, peerInfo) if shouldSendNewBlock(newBlock, peerInfo) => peer
-    }.toSet
+  def broadcastBlock(newBlock: BlockToBroadcast, handshakedPeers: Map[Peer, PeerInfo]): Unit = {
+    val peersWithoutBlock = handshakedPeers.filter { case (_, peerInfo) =>
+      shouldSendNewBlock(newBlock, peerInfo)
+    }
 
     broadcastNewBlock(newBlock, peersWithoutBlock)
 
     if (syncConfig.broadcastNewBlockHashes) {
       // NOTE: the usefulness of this message is debatable, especially in private networks
-      broadcastNewBlockHash(newBlock, peersWithoutBlock)
+      broadcastNewBlockHash(newBlock, peersWithoutBlock.keySet)
     }
   }
 
-  private def shouldSendNewBlock(newBlock: NewBlock, peerInfo: PeerInfo): Boolean =
+  private def shouldSendNewBlock(newBlock: BlockToBroadcast, peerInfo: PeerInfo): Boolean =
     newBlock.block.header.number > peerInfo.maxBlockNumber ||
       newBlock.chainWeight > peerInfo.chainWeight
 
-  private def broadcastNewBlock(newBlock: NewBlock, peers: Set[Peer]): Unit =
-    obtainRandomPeerSubset(peers).foreach { peer =>
-      etcPeerManager ! EtcPeerManagerActor.SendMessage(newBlock, peer.id)
+  private def broadcastNewBlock(newBlock: BlockToBroadcast, peers: Map[Peer, PeerInfo]): Unit =
+    obtainRandomPeerSubset(peers.keySet).foreach { peer =>
+      val message: MessageSerializable =
+        if (peers(peer).remoteStatus.protocolVersion == Versions.PV64) newBlock.as64 else newBlock.as63
+      etcPeerManager ! EtcPeerManagerActor.SendMessage(message, peer.id)
     }
 
-  private def broadcastNewBlockHash(newBlock: NewBlock, peers: Set[Peer]): Unit = peers.foreach { peer =>
+  private def broadcastNewBlockHash(newBlock: BlockToBroadcast, peers: Set[Peer]): Unit = peers.foreach { peer =>
     val newBlockHeader = newBlock.block.header
     val newBlockHashMsg = PV62.NewBlockHashes(Seq(BlockHash(newBlockHeader.hash, newBlockHeader.number)))
     etcPeerManager ! EtcPeerManagerActor.SendMessage(newBlockHashMsg, peer.id)
@@ -61,3 +65,22 @@ class BlockBroadcast(val etcPeerManager: ActorRef, syncConfig: SyncConfig) {
     Random.shuffle(peers).take(numberOfPeersToSend)
   }
 }
+
+object BlockBroadcast {
+
+  /**
+    * BlockToBroadcast was created to decouple block information from protocol new block messages
+    * (they are different versions of NewBlock msg)
+    */
+  case class BlockToBroadcast(block: Block, chainWeight: ChainWeight) {
+    def as63: CommonMessages.NewBlock = CommonMessages.NewBlock(block, chainWeight.totalDifficulty)
+    def as64: PV64.NewBlock = PV64.NewBlock(block, chainWeight)
+  }
+  object BlockToBroadcast {
+    def apply(block: CommonMessages.NewBlock): BlockToBroadcast =
+      BlockToBroadcast(block.block, ChainWeight.totalDifficultyOnly(block.totalDifficulty))
+
+    def apply(block: PV64.NewBlock): BlockToBroadcast =
+      BlockToBroadcast(block.block, block.chainWeight)
+  }
+}

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

@@ -7,13 +7,11 @@ import io.iohk.ethereum.consensus.validators.Validators
 import io.iohk.ethereum.db.storage.{AppStateStorage, FastSyncStateStorage}
 import io.iohk.ethereum.domain.Blockchain
 import io.iohk.ethereum.ledger.Ledger
-import io.iohk.ethereum.utils.BlockchainConfig
 import io.iohk.ethereum.utils.Config.SyncConfig
 
 class SyncController(
     appStateStorage: AppStateStorage,
     blockchain: Blockchain,
-    blockchainConfig: BlockchainConfig,
     fastSyncStateStorage: FastSyncStateStorage,
     ledger: Ledger,
     validators: Validators,
@@ -101,7 +99,6 @@ class SyncController(
         peerEventBus,
         ledger,
         blockchain,
-        blockchainConfig,
         syncConfig,
         ommersPool,
         pendingTransactionsManager,
@@ -121,7 +118,6 @@ object SyncController {
   def props(
       appStateStorage: AppStateStorage,
       blockchain: Blockchain,
-      blockchainConfig: BlockchainConfig,
       syncStateStorage: FastSyncStateStorage,
       ledger: Ledger,
       validators: Validators,
@@ -136,7 +132,6 @@ object SyncController {
       new SyncController(
         appStateStorage,
         blockchain,
-        blockchainConfig,
         syncStateStorage,
         ledger,
         validators,

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

@@ -1,8 +1,8 @@
 package io.iohk.ethereum.blockchain.sync.regular
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Props, Scheduler}
+import io.iohk.ethereum.blockchain.sync.BlockBroadcast.BlockToBroadcast
 import io.iohk.ethereum.blockchain.sync.{BlacklistSupport, BlockBroadcast, PeerListSupport}
-import io.iohk.ethereum.network.p2p.messages.CommonMessages.NewBlock
 import io.iohk.ethereum.utils.Config.SyncConfig
 
 class BlockBroadcasterActor(
@@ -26,8 +26,8 @@ class BlockBroadcasterActor(
 }
 object BlockBroadcasterActor {
   sealed trait BroadcasterMsg
-  case class BroadcastBlock(block: NewBlock) extends BroadcasterMsg
-  case class BroadcastBlocks(blocks: List[NewBlock]) extends BroadcasterMsg
+  case class BroadcastBlock(block: BlockToBroadcast) extends BroadcasterMsg
+  case class BroadcastBlocks(blocks: List[BlockToBroadcast]) extends BroadcasterMsg
 
   def props(
       broadcast: BlockBroadcast,

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

@@ -20,6 +20,8 @@ import io.iohk.ethereum.domain._
 import io.iohk.ethereum.network.PeerEventBusActor.PeerEvent.MessageFromPeer
 import io.iohk.ethereum.network.PeerEventBusActor.SubscriptionClassifier.MessageClassifier
 import io.iohk.ethereum.network.PeerEventBusActor.{PeerSelector, Subscribe, Unsubscribe}
+import io.iohk.ethereum.network.PeerId
+import io.iohk.ethereum.network.p2p.messages.{CommonMessages, PV64}
 import io.iohk.ethereum.network.p2p.messages.CommonMessages.NewBlock
 import io.iohk.ethereum.network.p2p.messages.PV62._
 import io.iohk.ethereum.network.p2p.messages.PV63.{GetNodeData, NodeData}
@@ -57,7 +59,7 @@ class BlockFetcher(
     BlockFetcherState.initial(importer, blockNr) |> fetchBlocks
     peerEventBus ! Subscribe(
       MessageClassifier(
-        Set(NewBlock.code63, NewBlock.code64, NewBlockHashes.code, BlockHeaders.code),
+        Set(NewBlock.code, NewBlockHashes.code, BlockHeaders.code),
         PeerSelector.AllPeers
       )
     )
@@ -192,45 +194,51 @@ class BlockFetcher(
       supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
 
       fetchBlocks(newState)
-    case MessageFromPeer(NewBlock(_, block, _), peerId) =>
-      val newBlockNr = block.number
-      val nextExpectedBlock = state.lastFullBlockNumber + 1
-
-      log.debug("Received NewBlock nr {}", newBlockNr)
-
-      // we're on top, so we can pass block directly to importer
-      if (newBlockNr == nextExpectedBlock && state.isOnTop) {
-        log.debug("Pass block directly to importer")
-        val newState = state.withPeerForBlocks(peerId, Seq(newBlockNr)).withKnownTopAt(newBlockNr)
-        state.importer ! OnTop
-        state.importer ! ImportNewBlock(block, peerId)
-        supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
-        context become started(newState)
-        // there are some blocks waiting for import but it seems that we reached top on fetch side so we can enqueue new block for import
-      } else if (newBlockNr == nextExpectedBlock && !state.isFetching && state.waitingHeaders.isEmpty) {
-        log.debug("Enqueue new block for import")
-        val newState = state.appendNewBlock(block, peerId)
-        supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
-        context become started(newState)
-        // waiting for some bodies but we don't have this header yet - at least we can use new block header
-      } else if (newBlockNr == state.nextToLastBlock && !state.isFetchingHeaders) {
-        log.debug("Waiting for bodies. Add only headers")
-        val newState = state.appendHeaders(List(block.header))
-        supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
-        fetchBlocks(newState)
-        // we're far from top
-      } else if (newBlockNr > nextExpectedBlock) {
-        log.debug("Far from top")
-        val newState = state.withKnownTopAt(newBlockNr)
-        supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
-        fetchBlocks(newState)
-      }
+    case MessageFromPeer(CommonMessages.NewBlock(block, _), peerId) =>
+      handleNewBlock(block, peerId, state)
+    case MessageFromPeer(PV64.NewBlock(block, _), peerId) =>
+      handleNewBlock(block, peerId, state)
     case BlockImportFailed(blockNr, reason) =>
       val (peerId, newState) = state.invalidateBlocksFrom(blockNr)
       peerId.foreach(id => peersClient ! BlacklistPeer(id, reason))
       fetchBlocks(newState)
   }
 
+  private def handleNewBlock(block: Block, peerId: PeerId, state: BlockFetcherState): Unit = {
+    val newBlockNr = block.number
+    val nextExpectedBlock = state.lastFullBlockNumber + 1
+
+    log.debug("Received NewBlock nr {}", newBlockNr)
+
+    // we're on top, so we can pass block directly to importer
+    if (newBlockNr == nextExpectedBlock && state.isOnTop) {
+      log.debug("Pass block directly to importer")
+      val newState = state.withPeerForBlocks(peerId, Seq(newBlockNr)).withKnownTopAt(newBlockNr)
+      state.importer ! OnTop
+      state.importer ! ImportNewBlock(block, peerId)
+      supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
+      context become started(newState)
+      // there are some blocks waiting for import but it seems that we reached top on fetch side so we can enqueue new block for import
+    } else if (newBlockNr == nextExpectedBlock && !state.isFetching && state.waitingHeaders.isEmpty) {
+      log.debug("Enqueue new block for import")
+      val newState = state.appendNewBlock(block, peerId)
+      supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
+      context become started(newState)
+      // waiting for some bodies but we don't have this header yet - at least we can use new block header
+    } else if (newBlockNr == state.nextToLastBlock && !state.isFetchingHeaders) {
+      log.debug("Waiting for bodies. Add only headers")
+      val newState = state.appendHeaders(List(block.header))
+      supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
+      fetchBlocks(newState)
+      // we're far from top
+    } else if (newBlockNr > nextExpectedBlock) {
+      log.debug("Far from top")
+      val newState = state.withKnownTopAt(newBlockNr)
+      supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
+      fetchBlocks(newState)
+    }
+  }
+
   private def handlePossibleTopUpdate(state: BlockFetcherState): Receive = {
     //by handling these type of messages, fetcher can received from network, fresh info about blocks on top
     //ex. After a successful handshake, fetcher will receive the info about the header of the peer best block