[ETCM-280] use peer capabilities (protocol version) when sending chec…

…kpoint-related messages

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.PeerDiscoveryManager.DiscoveredNodesInfo
 import io.iohk.ethereum.network.discovery.{DiscoveryConfig, Node}
-import io.iohk.ethereum.network.discovery.PeerDiscoveryManager.{DiscoveredNodesInfo}
 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)
@@ -227,7 +220,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/DumpChainActor.scala

@@ -6,26 +6,27 @@ import java.net.URI
 import akka.actor.{Actor, ActorRef, _}
 import akka.util.ByteString
 import io.iohk.ethereum.crypto.kec256
-import io.iohk.ethereum.domain.{BlockBody, BlockHeader, Receipt}
 import io.iohk.ethereum.domain.BlockHeaderImplicits._
-import io.iohk.ethereum.network.{Peer, PeerManagerActor}
+import io.iohk.ethereum.domain.{BlockBody, BlockHeader, Receipt}
+import io.iohk.ethereum.mpt.{BranchNode, ExtensionNode, HashNode, LeafNode, MptNode}
 import io.iohk.ethereum.network.PeerActor.SendMessage
+import io.iohk.ethereum.network.PeerEventBusActor.PeerEvent.MessageFromPeer
+import io.iohk.ethereum.network.PeerEventBusActor.SubscriptionClassifier.MessageClassifier
+import io.iohk.ethereum.network.PeerEventBusActor.{PeerSelector, Subscribe}
 import io.iohk.ethereum.network.PeerManagerActor.{GetPeers, Peers}
+import io.iohk.ethereum.network.p2p.messages.Codes
 import io.iohk.ethereum.network.p2p.messages.PV62._
-import io.iohk.ethereum.network.p2p.messages.PV63._
 import io.iohk.ethereum.network.p2p.messages.PV63.MptNodeEncoders._
+import io.iohk.ethereum.network.p2p.messages.PV63.ReceiptImplicits._
+import io.iohk.ethereum.network.p2p.messages.PV63._
+import io.iohk.ethereum.network.{Peer, PeerManagerActor}
+import io.iohk.ethereum.txExecTest.util.DumpChainActor._
 import org.bouncycastle.util.encoders.Hex
-import ReceiptImplicits._
-import io.iohk.ethereum.mpt.{BranchNode, ExtensionNode, HashNode, LeafNode, MptNode}
-import io.iohk.ethereum.network.PeerEventBusActor.PeerEvent.MessageFromPeer
-import io.iohk.ethereum.network.PeerEventBusActor.{PeerSelector, Subscribe}
-import io.iohk.ethereum.network.PeerEventBusActor.SubscriptionClassifier.MessageClassifier
 
 import scala.collection.immutable.HashMap
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration._
 import scala.language.postfixOps
-import DumpChainActor._
 
 /**
   * Actor used for obtaining all the blockchain data (blocks, receipts, nodes) from the blocks [startBlock, maxBlocks]
@@ -84,7 +85,7 @@ class DumpChainActor(
       peers.headOption.foreach { peer =>
         peerMessageBus ! Subscribe(
           MessageClassifier(
-            Set(BlockHeaders.code, BlockBodies.code, Receipts.code, NodeData.code),
+            Set(Codes.BlockHeadersCode, Codes.BlockBodiesCode, Codes.ReceiptsCode, Codes.NodeDataCode),
             PeerSelector.WithId(peer.id)
           )
         )

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/resources/application.conf

@@ -38,6 +38,8 @@ mantis {
 
   network {
     # Ethereum protocol version
+    # Supported versions:
+    # 63, 64 (experimental version which enables usage of messages with checkpointing information. In the future after ETCM-355, ETCM-356, it will be 66 probably)
     protocol-version = 63
 
     server-address {

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, ProtocolVersions}
+import io.iohk.ethereum.network.{EtcPeerManagerActor, Peer}
 import io.iohk.ethereum.utils.Config.SyncConfig
 
 import scala.util.Random
@@ -18,35 +20,39 @@ class BlockBroadcast(val etcPeerManager: ActorRef, syncConfig: SyncConfig) {
     * The hash of the block is sent to all of those peers while the block itself is only sent to
     * the square root of the total number of those peers, with the subset being obtained randomly.
     *
-    * @param newBlock, block to broadcast
+    * @param blockToBroadcast, 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(blockToBroadcast: BlockToBroadcast, handshakedPeers: Map[Peer, PeerInfo]): Unit = {
+    val peersWithoutBlock = handshakedPeers.filter { case (_, peerInfo) =>
+      shouldSendNewBlock(blockToBroadcast, peerInfo)
+    }
 
-    broadcastNewBlock(newBlock, peersWithoutBlock)
+    broadcastNewBlock(blockToBroadcast, peersWithoutBlock)
 
     if (syncConfig.broadcastNewBlockHashes) {
       // NOTE: the usefulness of this message is debatable, especially in private networks
-      broadcastNewBlockHash(newBlock, peersWithoutBlock)
+      broadcastNewBlockHash(blockToBroadcast, 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(blockToBroadcast: BlockToBroadcast, peers: Map[Peer, PeerInfo]): Unit =
+    obtainRandomPeerSubset(peers.keySet).foreach { peer =>
+      val message: MessageSerializable =
+        if (peers(peer).remoteStatus.protocolVersion == ProtocolVersions.PV64) blockToBroadcast.as64
+        else blockToBroadcast.as63
+      etcPeerManager ! EtcPeerManagerActor.SendMessage(message, peer.id)
     }
 
-  private def broadcastNewBlockHash(newBlock: NewBlock, 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)
+  private def broadcastNewBlockHash(blockToBroadcast: BlockToBroadcast, peers: Set[Peer]): Unit = peers.foreach {
+    peer =>
+      val newBlockHeader = blockToBroadcast.block.header
+      val newBlockHashMsg = PV62.NewBlockHashes(Seq(BlockHash(newBlockHeader.hash, newBlockHeader.number)))
+      etcPeerManager ! EtcPeerManagerActor.SendMessage(newBlockHashMsg, peer.id)
   }
 
   /**
@@ -61,3 +67,15 @@ 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)
+  }
+}

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

@@ -12,6 +12,7 @@ import io.iohk.ethereum.network.p2p.messages.PV62.{BlockBodies, BlockHeaders, Ge
 import io.iohk.ethereum.network.p2p.messages.PV63.{GetNodeData, GetReceipts, NodeData, Receipts}
 import io.iohk.ethereum.network.p2p.messages.PV63.MptNodeEncoders._
 import io.iohk.ethereum.network.EtcPeerManagerActor
+import io.iohk.ethereum.network.p2p.messages.Codes
 
 /**
   * BlockchainHost actor is in charge of replying to the peer's requests for blockchain data, which includes both
@@ -25,7 +26,8 @@ class BlockchainHostActor(
 ) extends Actor
     with ActorLogging {
 
-  private val requestMsgsCodes = Set(GetNodeData.code, GetReceipts.code, GetBlockBodies.code, GetBlockHeaders.code)
+  private val requestMsgsCodes =
+    Set(Codes.GetNodeDataCode, Codes.GetReceiptsCode, Codes.GetBlockBodiesCode, Codes.GetBlockHeadersCode)
   peerEventBusActor ! Subscribe(MessageClassifier(requestMsgsCodes, PeerSelector.AllPeers))
 
   override def receive: Receive = { case MessageFromPeer(message, peerId) =>

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

@@ -1,6 +1,7 @@
 package io.iohk.ethereum.blockchain.sync
 
 import java.time.Instant
+
 import akka.actor._
 import akka.util.ByteString
 import cats.data.NonEmptyList
@@ -20,11 +21,13 @@ import io.iohk.ethereum.domain._
 import io.iohk.ethereum.mpt.MerklePatriciaTrie
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.Peer
+import io.iohk.ethereum.network.p2p.messages.Codes
 import io.iohk.ethereum.network.p2p.messages.PV62._
 import io.iohk.ethereum.network.p2p.messages.PV63._
 import io.iohk.ethereum.utils.ByteStringUtils
 import io.iohk.ethereum.utils.Config.SyncConfig
 import org.bouncycastle.util.encoders.Hex
+
 import scala.annotation.tailrec
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration._
@@ -707,7 +710,7 @@ class FastSync(
           etcPeerManager,
           peerEventBus,
           requestMsg = GetReceipts(receiptsToGet),
-          responseMsgCode = Receipts.code
+          responseMsgCode = Codes.ReceiptsCode
         )
       )
 
@@ -728,7 +731,7 @@ class FastSync(
           etcPeerManager,
           peerEventBus,
           requestMsg = GetBlockBodies(blockBodiesToGet),
-          responseMsgCode = BlockBodies.code
+          responseMsgCode = Codes.BlockBodiesCode
         )
       )
 
@@ -753,7 +756,7 @@ class FastSync(
           etcPeerManager,
           peerEventBus,
           requestMsg = GetBlockHeaders(Left(syncState.bestBlockHeaderNumber + 1), limit, skip = 0, reverse = false),
-          responseMsgCode = BlockHeaders.code
+          responseMsgCode = Codes.BlockHeadersCode
         ),
         BlockHeadersHandlerName
       )

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

@@ -2,6 +2,7 @@ package io.iohk.ethereum.blockchain.sync
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Props, Scheduler}
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
+import io.iohk.ethereum.network.p2p.messages.Codes
 import io.iohk.ethereum.network.{Peer, PeerId}
 import io.iohk.ethereum.network.p2p.messages.PV62._
 import io.iohk.ethereum.network.p2p.messages.PV63.{GetNodeData, NodeData}
@@ -94,9 +95,9 @@ class PeersClient(
 
   private def responseMsgCode[RequestMsg <: Message](requestMsg: RequestMsg): Int =
     requestMsg match {
-      case _: GetBlockHeaders => BlockHeaders.code
-      case _: GetBlockBodies => BlockBodies.code
-      case _: GetNodeData => NodeData.code
+      case _: GetBlockHeaders => Codes.BlockHeadersCode
+      case _: GetBlockBodies => Codes.BlockBodiesCode
+      case _: GetNodeData => Codes.NodeDataCode
     }
 
   private def printStatus(requesters: Requesters): Unit = {

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

@@ -7,9 +7,11 @@ import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 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.p2p.messages.Codes
 import io.iohk.ethereum.network.p2p.messages.PV62.{BlockHeaders, GetBlockHeaders}
 import io.iohk.ethereum.network.{EtcPeerManagerActor, Peer, PeerId}
 import io.iohk.ethereum.utils.Config.SyncConfig
+
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.FiniteDuration
 
@@ -77,7 +79,7 @@ class PivotBlockSelector(
   ): Receive =
     handleCommonMessages orElse {
       case MessageFromPeer(blockHeaders: BlockHeaders, peerId) =>
-        peerEventBus ! Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peerId)))
+        peerEventBus ! Unsubscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peerId)))
         val updatedPeersToAsk = peersToAsk - peerId
         val targetBlockHeaderOpt =
           if (blockHeaders.headers.size != 1) None
@@ -165,7 +167,7 @@ class PivotBlockSelector(
   }
 
   private def obtainBlockHeaderFromPeer(peer: PeerId, blockNumber: BigInt): Unit = {
-    peerEventBus ! Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer)))
+    peerEventBus ! Subscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peer)))
     etcPeerManager ! EtcPeerManagerActor.SendMessage(
       GetBlockHeaders(Left(blockNumber), 1, 0, reverse = false),
       peer