Discovery: Filter out non 'eth' nodes and regularly update 'eth' field. (#3593)

* Discovery: Filter out non 'eth' nodes and regularly update 'eth' field.

* Rename UpdaterHook to something related to ForkId

* Use proper encoding for P2P ForkId

* Avoid using 'hash' for variable name

Author: jangko

execution_chain/common/common.nim

@@ -362,6 +362,9 @@ func forkId*(com: CommonRef, head: BlockNumber, time: EthTime): ForkID {.gcsafe.
   ## EIP 2364/2124
   com.forkIdCalculator.newID(head, time.uint64)
 
+func compatibleForkId*(com: CommonRef, id: ForkID): bool =
+  com.forkIdCalculator.compatible(id)
+
 func isEIP155*(com: CommonRef, number: BlockNumber): bool =
   com.config.eip155Block.isSome and number >= com.config.eip155Block.value
 

execution_chain/common/hardforks.nim

@@ -343,24 +343,43 @@ type
     byBlock: seq[uint64]
     byTime: seq[uint64]
     genesisCRC: uint32
+    cache: seq[ForkID]
 
 func newID*(calc: ForkIdCalculator, head, time: uint64): ForkID =
-  var hash = calc.genesisCRC
+  var crc = calc.genesisCRC
   for fork in calc.byBlock:
     if fork <= head:
-      # Fork already passed, checksum the previous hash and the fork number
-      hash = crc32(hash, fork.toBytesBE)
+      # Fork already passed, checksum the previous crc and the fork number
+      crc = crc32(crc, fork.toBytesBE)
       continue
-    return (hash, fork)
+    return (crc, fork)
 
   for fork in calc.byTime:
     if fork <= time:
-      # Fork already passed, checksum the previous hash and fork timestamp
-      hash = crc32(hash, fork.toBytesBE)
+      # Fork already passed, checksum the previous crc and fork timestamp
+      crc = crc32(crc, fork.toBytesBE)
       continue
-    return (hash, fork)
+    return (crc, fork)
 
-  (hash, 0'u64)
+  (crc, 0'u64)
+
+func compatible*(calc: var ForkIdCalculator, forkId: ForkID): bool =
+  if calc.cache.len == 0:
+    calc.cache = newSeqOfCap[ForkID](calc.byBlock.len + calc.byTime.len)
+    var crc = calc.genesisCRC
+    for fork in calc.byBlock:
+      crc = crc32(crc, fork.toBytesBE)
+      calc.cache.add( (crc, fork) )
+
+    for fork in calc.byTime:
+      crc = crc32(crc, fork.toBytesBE)
+      calc.cache.add( (crc, fork) )
+
+  for id in calc.cache:
+    if id == forkId:
+      return true
+
+  false
 
 func initForkIdCalculator*(map: ForkTransitionTable,
                            genesisCRC: uint32,

execution_chain/networking/chain_forkid.nim

@@ -0,0 +1,28 @@
+# nimbus-execution-client
+# Copyright (c) 2025 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms
+
+import
+  stew/endians2
+
+from eth/common/base import ForkID
+
+type
+  ChainForkId* = object
+    forkHash*: array[4, byte] # The RLP encoding must be exactly 4 bytes.
+    forkNext*: uint64         # The RLP encoding must be variable-length
+
+func to*(id: ChainForkId, _: type ForkID): ForkID =
+  (uint32.fromBytesBE(id.forkHash), id.forkNext)
+
+func to*(id: ForkID, _: type ChainForkId): ChainForkId =
+  ChainForkId(
+    forkHash: id.crc.toBytesBE,
+    forkNext: id.nextFork
+  )
+

execution_chain/networking/discoveryv5.nim

@@ -19,8 +19,9 @@ import
   eth/p2p/discoveryv5/protocol {.all.}
 
 export
-  Protocol, Node, Address, enr, newProtocol, open, close, seedTable, start, queryRandom, closeWait
-  
+  Protocol, Node, Address, enr, newProtocol, open, close, seedTable, start, queryRandom, closeWait,
+  updateRecord
+
 proc receiveV5*(d: Protocol, a: Address, packet: openArray[byte]): Result[void, cstring] =
   privateAccess(Protocol)
   privateAccess(PendingRequest)

execution_chain/networking/eth1_discovery.nim

@@ -15,9 +15,11 @@ import
   chronicles,
   results,
   metrics,
+  eth/common/base_rlp,
   ./discoveryv5,
   ./discoveryv4,
-  ./eth1_enr
+  ./eth1_enr,
+  ./chain_forkid
 
 export
   discoveryv4.NodeId,
@@ -37,9 +39,12 @@ type
   AddressV4 = discoveryv4.Address
   AddressV5 = discoveryv5.Address
 
+  CompatibleForkIdProc* = proc(id: ForkID): bool {.noSideEffect, raises: [].}
+
   Eth1Discovery* = ref object
     discv4: DiscV4
     discv5: DiscV5
+    compatibleForkId: CompatibleForkIdProc
 
 #------------------------------------------------------------------------------
 # Private functions
@@ -87,6 +92,23 @@ proc processClient(
     proto.discv5.receiveV5(addrv5, buf).isOkOr:
       debug "Discovery receive error", discv4=discv4.error, discv5=error
 
+func eligibleNode(proto: Eth1Discovery, rec: Record): bool =
+  # Filter out non `eth` node
+  let
+    bytes = rec.tryGet("eth", seq[byte]).valueOr:
+      return false
+
+  if proto.compatibleForkId.isNil:
+    # Allow all `eth` node to pass if there is no filter
+    return true
+
+  let
+    chainForkIds = try: rlp.decode(bytes, array[1, ChainForkId])
+              except RlpError: return false
+    chainForkId  = chainForkIds[0]
+
+  proto.compatibleForkId(chainForkId.to(ForkID))
+
 #------------------------------------------------------------------------------
 # Public functions
 #------------------------------------------------------------------------------
@@ -98,7 +120,8 @@ proc new*(
     bootstrapNodes: openArray[ENode],
     bindPort: Port,
     bindIp = IPv6_any(),
-    rng = newRng()
+    rng = newRng(),
+    compatibleForkId = CompatibleForkIdProc(nil)
 ): Eth1Discovery =
   let bootnodes = bootstrapNodes.to(enr.Record)
   Eth1Discovery(
@@ -120,7 +143,8 @@ proc new*(
       bindIp = bindIp,
       enrAutoUpdate = true,
       rng = rng
-    )
+    ),
+    compatibleForkId: compatibleForkId,
   )
 
 proc open*(
@@ -173,6 +197,8 @@ proc lookupRandomNode*(proto: Eth1Discovery, queue: AsyncQueue[NodeV4]) {.async:
   if proto.discv5.isNil.not:
     let nodes = await proto.discv5.queryRandom()
     for node in nodes:
+      if not proto.eligibleNode(node.record):
+        continue
       let v4 = node.to(NodeV4).valueOr:
         continue
       await queue.addLast(v4)
@@ -190,6 +216,16 @@ proc getRandomBootnode*(proto: Eth1Discovery): Opt[NodeV4] =
           return Opt.none(NodeV4)
       return Opt.some(newNode(enode))
 
+func updateForkID*(proto: Eth1Discovery, forkId: ForkID) =
+  # https://github.com/ethereum/devp2p/blob/bc76b9809a30e6dc5c8dcda996273f0f9bcf7108/enr-entries/eth.md
+  if proto.discv5.isNil.not:
+    let
+      list = [forkId.to(ChainForkId)]
+      bytes = rlp.encode(list)
+      kv = ("eth", bytes)
+    proto.discv5.updateRecord([kv]).isOkOr:
+      return
+
 proc closeWait*(proto: Eth1Discovery) {.async: (raises: []).} =
   privateAccess(DiscV4)
   if proto.discv4.isNil.not and proto.discv5.isNil:

execution_chain/networking/p2p.nim

@@ -19,11 +19,16 @@ import
   ./eth1_discovery
 
 export
-  p2p_types, rlpx, enode
+  p2p_types, rlpx, enode, ForkIdProc, CompatibleForkIdProc
 
 logScope:
   topics = "p2p"
 
+type
+  ForkIdProcs* = object
+    forkId*: ForkIdProc
+    compatibleForkId*: CompatibleForkIdProc
+
 proc newEthereumNode*(
     keys: KeyPair,
     address: Address,
@@ -34,14 +39,15 @@ proc newEthereumNode*(
     bindUdpPort: Port,
     bindTcpPort: Port,
     bindIp = IPv6_any(),
-    rng = newRng()): EthereumNode =
+    rng = newRng(),
+    forkIdProcs = ForkIdProcs()): EthereumNode =
 
   if rng == nil: # newRng could fail
     raiseAssert "Cannot initialize RNG"
 
   let
     discovery = Eth1Discovery.new(
-      keys.seckey, address, bootstrapNodes, bindUdpPort, bindIp, rng)
+      keys.seckey, address, bootstrapNodes, bindUdpPort, bindIp, rng, forkIdProcs.compatibleForkId)
     node = EthereumNode(
       keys: keys,
       networkId: networkId,
@@ -53,7 +59,7 @@ proc newEthereumNode*(
       rng: rng,
     )
   node.peerPool = newPeerPool[EthereumNode](
-    node, discovery, minPeers = minPeers)
+    node, discovery, minPeers = minPeers, forkId = forkIdProcs.forkId)
   node
 
 proc processIncoming(server: StreamServer,

execution_chain/networking/peer_pool.nim

@@ -18,7 +18,9 @@ import
   ./p2p_metrics,
   ./[eth1_discovery, p2p_peers]
 
-export sets, tables
+from eth/common/base import ForkID
+
+export sets, tables, CompatibleForkIdProc
 
 logScope:
   topics = "p2p peer_pool"
@@ -30,6 +32,8 @@ type
 
   WorkerFuture = Future[void].Raising([CancelledError])
 
+  ForkIdProc* = proc(): ForkID {.noSideEffect, raises: [].}
+
   # Usually Network generic param is instantiated with EthereumNode
   PeerPoolRef*[Network] = ref object
     network: Network
@@ -40,6 +44,10 @@ type
     running: bool
     discovery: Eth1Discovery
     workers: seq[WorkerFuture]
+    forkId: ForkIdProc
+    lastForkId: ForkID
+    connectTimer: Future[void].Raising([CancelledError])
+    updateTimer: Future[void].Raising([CancelledError])
     connectingNodes*: HashSet[Node]
     connectedNodes*: Table[Node, PeerRef[Network]]
     observers*: Table[int, PeerObserverRef[Network]]
@@ -52,6 +60,7 @@ type
 const
   lookupInterval = 5
   connectLoopSleep = chronos.milliseconds(2000)
+  updateLoopSleep = chronos.seconds(15)
   maxConcurrentConnectionRequests = 40
   sleepBeforeTryingARandomBootnode = chronos.milliseconds(3000)
 
@@ -188,23 +197,55 @@ proc maybeConnectToMorePeers(p: PeerPoolRef) {.async: (raises: [CancelledError])
     return
   await p.connectToNode(n)
 
+func updateForkID(p: PeerPoolRef) =
+  if p.forkId.isNil:
+    return
+
+  let forkId = p.forkId()
+  if p.lastForkId == forkId:
+    return
+
+  p.discovery.updateForkID(forkId)
+  p.lastForkId = forkId
+
 proc run(p: PeerPoolRef) {.async: (raises: [CancelledError]).} =
   trace "Running PeerPool..."
 
+  # initial cycle
+  p.updateForkID()
   await p.discovery.start()
+  await p.maybeConnectToMorePeers()
+
   p.running = true
   while p.running:
     debug "Amount of peers", amount = p.connectedNodes.len()
-    await p.maybeConnectToMorePeers()
-    await sleepAsync(connectLoopSleep)
+
+    # Create or replenish timer
+    if p.connectTimer.isNil or p.connectTimer.finished:
+      p.connectTimer = sleepAsync(connectLoopSleep)
+
+    if p.updateTimer.isNil or p.updateTimer.finished:
+      p.updateTimer = sleepAsync(updateLoopSleep)
+
+    let
+      res = await one(p.connectTimer, p.updateTimer)
+
+    if res == p.connectTimer:
+      await p.maybeConnectToMorePeers()
+
+    if res == p.updateTimer:
+      p.updateForkID()
 
 #------------------------------------------------------------------------------
 # Private functions
 #------------------------------------------------------------------------------
 
 func newPeerPool*[Network](
     network: Network,
-    discovery: Eth1Discovery, minPeers = 10): PeerPoolRef[Network] =
+    discovery: Eth1Discovery,
+    minPeers = 10,
+    forkId = ForkIdProc(nil),
+    ): PeerPoolRef[Network] =
   new result
   result.network = network
   result.minPeers = minPeers
@@ -213,6 +254,7 @@ func newPeerPool*[Network](
   result.connectedNodes = initTable[Node, PeerRef[Network]]()
   result.connectingNodes = initHashSet[Node]()
   result.observers = initTable[int, PeerObserverRef[Network]]()
+  result.forkId = forkId
 
 proc addObserver*(p: PeerPoolRef, observerId: int, observer: PeerObserverRef) =
   doAssert(observerId notin p.observers)

execution_chain/nimbus_execution_client.nim

@@ -104,15 +104,30 @@ proc setupP2P(nimbus: NimbusNode, conf: NimbusConf,
       if extPorts.isSome:
         (address.tcpPort, address.udpPort) = extPorts.get()
 
-  let bootstrapNodes = conf.getBootNodes()
+  let
+    bootstrapNodes = conf.getBootNodes()
+    fc = nimbus.fc
+
+  func forkIdProc(): ForkID {.raises: [].} =
+    let header = fc.latestHeader()
+    com.forkId(header.number, header.timestamp)
+
+  func compatibleForkIdProc(id: ForkID): bool {.raises: [].} =
+    com.compatibleForkId(id)
+    
+  let forkIdProcs = ForkIdProcs(
+    forkId: forkIdProc,
+    compatibleForkId: compatibleForkIdProc,
+  )
 
   nimbus.ethNode = newEthereumNode(
     keypair, address, conf.networkId, conf.agentString,
     minPeers = conf.maxPeers,
     bootstrapNodes = bootstrapNodes,
     bindUdpPort = conf.udpPort, bindTcpPort = conf.tcpPort,
     bindIp = conf.listenAddress,
-    rng = nimbus.ctx.rng)
+    rng = nimbus.ctx.rng,
+    forkIdProcs = forkIdProcs)
 
   # Add protocol capabilities
   nimbus.wire = nimbus.ethNode.addEthHandlerCapability(nimbus.txPool)