Support multiple wire protocol version and implement eth/68 + eth/69 (#3283)

* Remove old wire protocol implementation

* eth68 status isolated

* eth69 preparation

* Fix typo

* Register protocol

* Add BlockRangeUpdate

* Use new receipt format for eth69

* Fix tests

* Update wire protocol setup

* Update syncer addObserver

* Update peer observer

* Handle blockRangeUpdate using peer state

* Add receipt69 roundtrip test

* Replace Receipt69 with StoredReceipt from nim-eth

* Bump nim-eth

* Bump nim-eth to master branch

Author: jangko

execution_chain/networking/p2p_types.nim

@@ -78,7 +78,7 @@ type
   PeerObserver* = object
     onPeerConnected*: proc(p: Peer) {.gcsafe, raises: [].}
     onPeerDisconnected*: proc(p: Peer) {.gcsafe, raises: [].}
-    protocol*: ProtocolInfo
+    protocols*: seq[ProtocolInfo]
 
   Capability* = object
     name*: string

execution_chain/networking/peer_pool.nim

@@ -87,7 +87,7 @@ proc addObserver*(p: PeerPool, observerId: int, observer: PeerObserver) =
   p.observers[observerId] = observer
   if not observer.onPeerConnected.isNil:
     for peer in p.connectedNodes.values:
-      if observer.protocol.isNil or peer.supports(observer.protocol):
+      if observer.protocols.len == 0 or peer.supports(observer.protocols):
         observer.onPeerConnected(peer)
 
 func delObserver*(p: PeerPool, observerId: int) =
@@ -99,19 +99,15 @@ proc addObserver*(p: PeerPool, observerId: ref, observer: PeerObserver) =
 func delObserver*(p: PeerPool, observerId: ref) =
   p.delObserver(cast[int](observerId))
 
-template setProtocol*(observer: PeerObserver, Protocol: type) =
-  observer.protocol = Protocol.protocolInfo
+template addProtocol*(observer: PeerObserver, Protocol: type) =
+  observer.protocols.add Protocol.protocolInfo
 
 proc stopAllPeers(p: PeerPool) {.async.} =
   debug "Stopping all peers ..."
   # TODO: ...
   # await asyncio.gather(
   #   *[peer.stop() for peer in self.connected_nodes.values()])
 
-# async def stop(self) -> None:
-#   self.cancel_token.trigger()
-#   await self.stop_all_peers()
-
 proc connect(p: PeerPool, remote: Node): Future[Peer] {.async.} =
   ## Connect to the given remote and return a Peer instance when successful.
   ## Returns nil if the remote is unreachable, times out or is useless.
@@ -166,10 +162,10 @@ proc addPeer*(pool: PeerPool, peer: Peer) {.gcsafe.} =
   doAssert(peer.remote notin pool.connectedNodes)
   pool.connectedNodes[peer.remote] = peer
   rlpx_connected_peers.inc()
-  for o in pool.observers.values:
-    if not o.onPeerConnected.isNil:
-      if o.protocol.isNil or peer.supports(o.protocol):
-        o.onPeerConnected(peer)
+  for observer in pool.observers.values:
+    if not observer.onPeerConnected.isNil:
+      if observer.protocols.len == 0 or peer.supports(observer.protocols):
+        observer.onPeerConnected(peer)
 
 proc connectToNode*(p: PeerPool, n: Node) {.async.} =
   let peer = await p.connect(n)

execution_chain/networking/rlpx.nim

@@ -365,6 +365,11 @@ proc supports*(peer: Peer, Protocol: type): bool =
   ## Checks whether a Peer supports a particular protocol
   peer.supports(Protocol.protocolInfo)
 
+proc supports*(peer: Peer, protos: openArray[ProtocolInfo]): bool =
+  for proto in protos:
+    if peer.supports(proto):
+      return true
+
 template perPeerMsgId(peer: Peer, MsgType: type): uint64 =
   perPeerMsgIdImpl(peer, MsgType.msgProtocol.protocolInfo, MsgType.msgId)
 
@@ -981,7 +986,7 @@ proc removePeer(network: EthereumNode, peer: Peer) =
     if not peer.dispatcher.isNil:
       for observer in network.peerPool.observers.values:
         if not observer.onPeerDisconnected.isNil:
-          if observer.protocol.isNil or peer.supports(observer.protocol):
+          if observer.protocols.len == 0 or peer.supports(observer.protocols):
             observer.onPeerDisconnected(peer)
 
 proc callDisconnectHandlers(
@@ -1557,6 +1562,26 @@ template rlpxWithFutureHandler*(PROTO: distinct type;
     resolveResponseFuture(peer,
       perPeerMsgId, addr(packet), reqId)
 
+template rlpxWithFutureHandler*(PROTO: distinct type;
+                        MSGTYPE: distinct type;
+                        PROTYPE: distinct type;
+                        msgId: static[uint64];
+                        peer: Peer;
+                        data: Rlp,
+                        fields: untyped): untyped =
+  wrapRlpxWithPacketException(MSGTYPE, peer):
+    var
+      rlp = data
+      packet: MSGTYPE
+
+    tryEnterList(rlp)
+    let
+      reqId = read(rlp, uint64)
+      perPeerMsgId = msgIdImpl(PROTO, peer, msgId)
+    checkedRlpFields(peer, rlp, packet, fields)
+    var proType = packet.to(PROTYPE)
+    resolveResponseFuture(peer,
+      perPeerMsgId, addr(proType), reqId)
 
 proc nextMsg*(PROTO: distinct type,
               peer: Peer,
@@ -1595,6 +1620,9 @@ func initResponder*(peer: Peer, reqId: uint64): Responder =
 template state*(response: Responder, PROTO: type): auto =
   state(response.peer, PROTO)
 
+template supports*(response: Responder, Protocol: type): bool =
+  response.peer.supports(Protocol.protocolInfo)
+
 template networkState*(response: Responder, PROTO: type): auto =
   networkState(response.peer, PROTO)
 
@@ -1618,6 +1646,9 @@ template defineProtocol*(PROTO: untyped,
   template NetworkState*(_: type PROTO): type =
     networkState
 
+  template protocolVersion*(_: type PROTO): int =
+    version
+
   func initProtocol*(_: type PROTO): auto =
     initProtocol(rlpxName,
       version,

execution_chain/sync/beacon/worker/start_stop.nim

@@ -94,8 +94,15 @@ proc startBuddy*(buddy: BeaconBuddyRef): bool =
   let
     ctx = buddy.ctx
     peer = buddy.peer
-  if peer.supports(wire_protocol.eth) and
-     peer.state(wire_protocol.eth).initialized:
+
+  if peer.supports(eth69) and
+     peer.state(eth69).initialized:
+    ctx.pool.nBuddies.inc
+    buddy.initHdrProcErrors()
+    return true
+
+  if peer.supports(eth68) and
+     peer.state(eth68).initialized:
     ctx.pool.nBuddies.inc
     ctx.pool.blkLastSlowPeer = Opt.none(Hash)
     buddy.initHdrProcErrors()

execution_chain/sync/peers.nim

@@ -92,7 +92,8 @@ proc setupManager(pm: PeerManagerRef, enodes: openArray[ENode]) =
       pm.state = Running
       pm.reconnectFut = pm.runReconnectLoop()
 
-  po.setProtocol eth
+  po.addProtocol eth68
+  po.addProtocol eth69
   pm.pool.addObserver(pm, po)
 
   for enode in enodes:

execution_chain/sync/sync_sched.nim

@@ -88,7 +88,8 @@ import
   chronos,
   ../networking/[p2p, peer_pool],
   stew/keyed_queue,
-  ./sync_desc
+  ./sync_desc,
+  ./wire_protocol
 
 type
   ActiveBuddies[S,W] = ##\
@@ -503,8 +504,10 @@ proc startSync*[S,W](dsc: RunnerSyncRef[S,W]): bool =
         onPeerDisconnected: proc(p: Peer) {.gcsafe.} =
           dsc.onPeerDisconnected(p))
 
-      po.setProtocol eth
+      po.addProtocol eth68
+      po.addProtocol eth69
       dsc.pool.addObserver(dsc, po)
+
       asyncSpawn dsc.tickerLoop()
       return true
 

execution_chain/sync/wire_protocol.nim

@@ -18,6 +18,3 @@ export
   responder,
   types,
   setup
-
-type
-  eth* = eth68

execution_chain/sync/wire_protocol/handler.nim

@@ -44,14 +44,14 @@ proc new*(_: type EthWireRef,
 # Public functions: eth wire protocol handlers
 # ------------------------------------------------------------------------------
 
-proc getStatus*(ctx: EthWireRef): EthState =
+proc getStatus68*(ctx: EthWireRef): Eth68State =
   let
     com = ctx.chain.com
     bestBlock = ctx.chain.latestHeader
     txFrame = ctx.chain.baseTxFrame
     forkId = com.forkId(bestBlock.number, bestBlock.timestamp)
 
-  EthState(
+  Eth68State(
     totalDifficulty: txFrame.headTotalDifficulty,
     genesisHash: com.genesisHash,
     bestBlockHash: bestBlock.computeBlockHash,
@@ -60,6 +60,23 @@ proc getStatus*(ctx: EthWireRef): EthState =
       forkNext: forkId.nextFork
   ))
 
+proc getStatus69*(ctx: EthWireRef): Eth69State =
+  let
+    com = ctx.chain.com
+    bestBlock = ctx.chain.latestHeader
+    forkId = com.forkId(bestBlock.number, bestBlock.timestamp)
+
+  Eth69State(
+    genesisHash: com.genesisHash,
+    forkId: ChainForkId(
+      forkHash: forkId.crc.toBytesBE,
+      forkNext: forkId.nextFork
+    ),
+    earliest: 0,
+    latest: bestBlock.number,
+    latestHash: bestBlock.computeBlockHash,
+  )
+
 proc getReceipts*(ctx: EthWireRef,
                   hashes: openArray[Hash32]):
                     seq[seq[Receipt]] =
@@ -80,6 +97,26 @@ proc getReceipts*(ctx: EthWireRef,
 
   move(list)
 
+proc getStoredReceipts*(ctx: EthWireRef,
+                  hashes: openArray[Hash32]):
+                    seq[seq[StoredReceipt]] =
+  var
+    list: seq[seq[StoredReceipt]]
+    totalBytes = 0
+
+  for blockHash in hashes:
+    var receiptList = ctx.chain.receiptsByBlockHash(blockHash).valueOr:
+      continue
+
+    totalBytes += getEncodedLength(receiptList)
+    list.add(receiptList.to(seq[StoredReceipt]))
+
+    if list.len >= MAX_RECEIPTS_SERVE or
+       totalBytes > SOFT_RESPONSE_LIMIT:
+      break
+
+  move(list)
+
 proc getPooledTransactions*(ctx: EthWireRef,
                      hashes: openArray[Hash32]):
                        seq[PooledTransaction] =
@@ -127,7 +164,7 @@ proc getBlockBodies*(ctx: EthWireRef,
       if blk.header.number > ctx.chain.portal.limit:
         trace "handlers.getBlockBodies: blockBody older than expiry limit", blockHash
         continue
-    
+
     totalBytes += getEncodedLength(blk.body)
     list.add blk.body