fusaka devnet 2 branch (#7269)

* some clarifications

* push event topic update for data column sidecar

* prevent inhibiting validator custody

* increase vcus poll interval

* not so cool devnet hack for now

* some validator custody and status v fixes

* few more fixes to event stream and vcus

* fixes

* few more changes

* clarifications regarding blob parameters

* revert reqman hack and remove assertion

* add more logging and reduce getblobs timeout

* cancel blob loop post fulu fork epoch

* rework validator custody counting logic and remove another assertion

* rman hack 2

* clarifications in validator custody logic

* reduce validator custody polling duration

* have validator custody detection and custody backfill on separate loops

* oops

* added extra logging for clarity

* use total attached balance instead of active balance

* some more rework on getBlobsV2

* other fixes

* off vcus for supernodes

* make the BN pass min DA requirements to catch missing blocks

* some fixes to rman

* reword peer filtering and scoring

* revise score

* bump up parallel requests as there are more number of cancellations

* omit minDA criteria

* bump down ll requests for supernodes

* fix beacon block broadcast not using BPO forkdigests (#7285)

* gate status vx

* patch blob schedule

* oops bug

---------

Co-authored-by: tersec <[email protected]>

Author: agnxsh

beacon_chain/consensus_object_pools/block_pools_types.nim

@@ -137,7 +137,7 @@ type
 
     erSlot*: Slot
       ## Earliest refilled slot is the earliest slot at which excess
-      ## DataColumnSidecar downloading starts, if erSlot = GENESIS_SLOT
+      ## DataColumnSidecar downloading finishes, if erSlot = GENESIS_SLOT
       ## we can deduce that validator custody is inactive.
 
     validatorMonitor*: ref ValidatorMonitor
@@ -411,14 +411,13 @@ func earliestAvailableSlot*(dag: ChainDAGRef): Slot =
       dag.backfill.slot != GENESIS_SLOT:
     # When the BN is backfilling, backfill slot is the earliest
     # persisted block.
-    dag.eaSlot = dag.backfill.slot
     dag.backfill.slot
   elif dag.erSlot != GENESIS_SLOT:
-    dag.eaSlot = dag.erSlot
+    # This indicates column filling due to validator custody
+    # is in progress
     dag.erSlot
   else:
     # When the BN has backfilled, tail moves progressively.
-    dag.eaSlot = dag.tail.slot
     dag.tail.slot
 
 template epoch*(e: EpochRef): Epoch = e.key.epoch

beacon_chain/consensus_object_pools/block_quarantine.nim

@@ -397,6 +397,14 @@ func popColumnless*(
   else:
     Opt.none(ForkedSignedBeaconBlock)
 
+func getColumnless*(
+    quarantine: var Quarantine,
+    root: Eth2Digest): Opt[ForkedSignedBeaconBlock] =
+  try:
+    Opt.some(quarantine.columnless[root])
+  except KeyError:
+    Opt.none(ForkedSignedBeaconBlock)
+
 iterator peekBlobless*(quarantine: var Quarantine): ForkedSignedBeaconBlock =
   for k, v in quarantine.blobless.mpairs():
     yield v

beacon_chain/el/el_manager.nim

@@ -66,7 +66,7 @@ const
   # https://github.com/ethereum/execution-apis/blob/v1.0.0-beta.4/src/engine/shanghai.md#request-2
   GETPAYLOAD_TIMEOUT = 1.seconds
 
-  GETBLOBS_TIMEOUT = 200.milliseconds
+  GETBLOBS_TIMEOUT = 100.milliseconds
 
   connectionStateChangeHysteresisThreshold = 15
     ## How many unsuccesful/successful requests we must see

beacon_chain/gossip_processing/block_processor.nim

@@ -979,7 +979,7 @@ proc storeBlock(
               MsgSource.gossip, quarantined, Opt.none(BlobSidecars),
               cres)
           else:
-            discard self.consensusManager.quarantine[].addBlobless(
+            discard self.consensusManager.quarantine[].addColumnless(
               dag.finalizedHead.slot, forkyBlck)
       elif typeof(forkyBlck).kind >= ConsensusFork.Deneb and
           typeof(forkyBlck).kind < ConsensusFork.Fulu:

beacon_chain/gossip_processing/eth2_processor.nim

@@ -354,11 +354,10 @@ proc processBlobSidecar*(
 
 proc validateDataColumnSidecarFromEL*(
     self: ref Eth2Processor,
-    block_root: Eth2Digest):
-    Future[ValidationRes]
+    block_root: Eth2Digest)
     {.async: (raises: [CancelledError]).} =
   let elManager = self.blockProcessor[].consensusManager.elManager
-  if (let o = self.quarantine[].popColumnless(block_root); o.isSome):
+  if (let o = self.quarantine[].getColumnless(block_root); o.isSome):
     let columnless = o.unsafeGet()
     withBlck(columnless):
       when consensusFork >= ConsensusFork.Fulu:
@@ -387,28 +386,12 @@ proc validateDataColumnSidecarFromEL*(
                   blobsEl.mapIt(kzg.KzgBlob(bytes: it.blob.data)),
                   flat_proof)
 
-            # Pop out the column sidecars as we have all columns from the EL
-            discard self.dataColumnQuarantine[].popSidecars(block_root,
-                                                            forkyBlck)
-
-            let end_time = Moment.now()
-            debug "Time taken to get 100% response from EL and bypass blob gossip validation",
-                  time_taken = end_time - start_time
-            debug "Pulled blobs from EL, bypassing blob gossip validation",
-                  blobs_from_el = blobsEl.len
-            self.blockProcessor[].enqueueBlock(
-              MsgSource.gossip, columnless,
-              Opt.none(BlobSidecars),
-              Opt.some(recovered_columns.mapIt(newClone it)))
-            return ok()
-
-          else:
-            discard self.quarantine[].addColumnless(
-              self.dag.finalizedHead.slot, forkyBlck)
-      else:
-          raiseAssert "Could not have been added as columnless"
-  else:
-    return errIgnore ("Could not pull blobs and proofs from EL")
+            # Send notification to event stream
+            # and add these columns to column quarantine
+            for col in recovered_columns:
+
+              if col.index in self.dataColumnQuarantine[].custodyColumns:
+                self.dataColumnQuarantine[].put(block_root, newClone(col))
 
 proc processDataColumnSidecar*(
     self: ref Eth2Processor, src: MsgSource,
@@ -417,11 +400,7 @@ proc processDataColumnSidecar*(
   template block_header: untyped = dataColumnSidecar.signed_block_header.message
   let block_root = hash_tree_root(block_header)
 
-  let vEL =
-    await self.validateDataColumnSidecarFromEL(block_root)
-
-  if vEL.isOk():
-    return vEL
+  await self.validateDataColumnSidecarFromEL(block_root)
 
   let
     wallTime = self.getCurrentBeaconTime()
@@ -446,7 +425,6 @@ proc processDataColumnSidecar*(
 
   debug "Data column validated, putting data column in quarantine"
   self.dataColumnQuarantine[].put(block_root, newClone(dataColumnSidecar))
-  self.dag.db.putDataColumnSidecar(dataColumnSidecar)
 
   if (let o = self.quarantine[].popColumnless(block_root); o.isSome):
     let columnless = o.unsafeGet()

beacon_chain/networking/eth2_network.nim

@@ -2793,45 +2793,10 @@ proc broadcastAggregateAndProof*(
   node.broadcast(topic, proof)
 
 proc broadcastBeaconBlock*(
-    node: Eth2Node, blck: phase0.SignedBeaconBlock):
+    node: Eth2Node, blck: SomeForkySignedBeaconBlock):
     Future[SendResult] {.async: (raises: [CancelledError], raw: true).} =
-  let topic = getBeaconBlocksTopic(node.forkDigests.phase0)
-  node.broadcast(topic, blck)
-
-proc broadcastBeaconBlock*(
-    node: Eth2Node, blck: altair.SignedBeaconBlock):
-    Future[SendResult] {.async: (raises: [CancelledError], raw: true).} =
-  let topic = getBeaconBlocksTopic(node.forkDigests.altair)
-  node.broadcast(topic, blck)
-
-proc broadcastBeaconBlock*(
-    node: Eth2Node, blck: bellatrix.SignedBeaconBlock):
-    Future[SendResult] {.async: (raises: [CancelledError], raw: true).} =
-  let topic = getBeaconBlocksTopic(node.forkDigests.bellatrix)
-  node.broadcast(topic, blck)
-
-proc broadcastBeaconBlock*(
-    node: Eth2Node, blck: capella.SignedBeaconBlock):
-    Future[SendResult] {.async: (raises: [CancelledError], raw: true).} =
-  let topic = getBeaconBlocksTopic(node.forkDigests.capella)
-  node.broadcast(topic, blck)
-
-proc broadcastBeaconBlock*(
-    node: Eth2Node, blck: deneb.SignedBeaconBlock):
-    Future[SendResult] {.async: (raises: [CancelledError], raw: true).} =
-  let topic = getBeaconBlocksTopic(node.forkDigests.deneb)
-  node.broadcast(topic, blck)
-
-proc broadcastBeaconBlock*(
-    node: Eth2Node, blck: electra.SignedBeaconBlock):
-    Future[SendResult] {.async: (raises: [CancelledError], raw: true).} =
-  let topic = getBeaconBlocksTopic(node.forkDigests.electra)
-  node.broadcast(topic, blck)
-
-proc broadcastBeaconBlock*(
-    node: Eth2Node, blck: fulu.SignedBeaconBlock):
-    Future[SendResult] {.async: (raises: [CancelledError], raw: true).} =
-  let topic = getBeaconBlocksTopic(node.forkDigests.fulu)
+  let topic = getBeaconBlocksTopic(
+    node.forkDigestAtEpoch(blck.message.slot.epoch))
   node.broadcast(topic, blck)
 
 proc broadcastBlobSidecar*(

beacon_chain/networking/peer_protocol.nim

@@ -76,7 +76,7 @@ func forkDigestAtEpoch(state: PeerSyncNetworkState,
   state.forkDigests[].atEpoch(epoch, state.cfg)
 
 # https://github.com/ethereum/consensus-specs/blob/v1.5.0-beta.0/specs/phase0/p2p-interface.md#status
-proc getCurrentStatus(state: PeerSyncNetworkState): StatusMsg =
+proc getCurrentStatusV1(state: PeerSyncNetworkState): StatusMsg =
   let
     dag = state.dag
     wallSlot = state.getBeaconTime().slotOrZero
@@ -166,13 +166,15 @@ proc checkStatusMsg(state: PeerSyncNetworkState, status: StatusMsg | StatusMsgV2
       # apparently don't use spec ZERO_HASH as of this writing
       if not (status.finalizedRoot in [state.genesisBlockRoot, ZERO_HASH]):
         return err("peer following different finality")
-
   ok()
 
-proc handleStatus(peer: Peer,
-                  state: PeerSyncNetworkState,
-                  theirStatus: StatusMsg): Future[bool] {.async: (raises: [CancelledError]).}
+proc handleStatusV1(peer: Peer,
+                    state: PeerSyncNetworkState,
+                    theirStatus: StatusMsg): Future[bool] {.async: (raises: [CancelledError]).}
 
+proc handleStatusV2(peer: Peer,
+                    state: PeerSyncNetworkState,
+                    theirStatus: StatusMsgV2): Future[bool] {.async: (raises: [CancelledError]).}
 
 {.pop.} # TODO fix p2p macro for raises
 
@@ -195,27 +197,55 @@ p2pProtocol PeerSync(version = 1,
     #      need a dedicated flow in libp2p that resolves the race conditions -
     #      this needs more thinking around the ordering of events and the
     #      given incoming flag
+
     let
-      ourStatus = peer.networkState.getCurrentStatus()
-      theirStatus = await peer.status(ourStatus, timeout = RESP_TIMEOUT_DUR)
-      ourStatusV2 = peer.networkState.getCurrentStatusV2()
+      remoteFork = peer.networkState.getBeaconTime().slotOrZero.epoch()
+
+    if remoteFork >= peer.networkState.cfg.FULU_FORK_EPOCH:
+      let
+        ourStatus = peer.networkState.getCurrentStatusV2()
+        theirStatus =
+          await peer.statusV2(ourStatus, timeout = RESP_TIMEOUT_DUR)
+
+      if theirStatus.isOk:
+        discard await peer.handleStatusV2(peer.networkState, theirStatus.get())
+        peer.updateAgent()
+      else:
+        debug "Status response not received in time",
+              peer, errorKind = theirStatus.error.kind
+        await peer.disconnect(FaultOrError)
 
-    if theirStatus.isOk:
-      discard await peer.handleStatus(peer.networkState, theirStatus.get())
-      peer.updateAgent()
     else:
-      debug "Status response not received in time",
-            peer, errorKind = theirStatus.error.kind
-      await peer.disconnect(FaultOrError)
-
-  proc status(peer: Peer,
-              theirStatus: StatusMsg,
-              response: SingleChunkResponse[StatusMsg])
-    {.async, libp2pProtocol("status", 1).} =
-    let ourStatus = peer.networkState.getCurrentStatus()
-    trace "Sending status message", peer = peer, status = ourStatus
+      let
+        ourStatus = peer.networkState.getCurrentStatusV1()
+        theirStatus =
+          await peer.statusV1(ourStatus, timeout = RESP_TIMEOUT_DUR)
+
+      if theirStatus.isOk:
+        discard await peer.handleStatusV1(peer.networkState, theirStatus.get())
+        peer.updateAgent()
+      else:
+        debug "Status response not received in time",
+              peer, errorKind = theirStatus.error.kind
+        await peer.disconnect(FaultOrError)
+
+  proc statusV1(peer: Peer,
+                theirStatus: StatusMsg,
+                response: SingleChunkResponse[StatusMsg])
+      {.async, libp2pProtocol("status", 1).} =
+    let ourStatus = peer.networkState.getCurrentStatusV1()
+    trace "Sending status (v1)", peer = peer, status = ourStatus
     await response.send(ourStatus)
-    discard await peer.handleStatus(peer.networkState, theirStatus)
+    discard await peer.handleStatusV1(peer.networkState, theirStatus)
+
+  proc statusV2(peer: Peer,
+                theirStatus: StatusMsgV2,
+                response: SingleChunkResponse[StatusMsgV2])
+      {.async, libp2pProtocol("status", 2).} =
+    let ourStatus = peer.networkState.getCurrentStatusV2()
+    trace "Sending status (v2)", peer = peer, status = ourStatus
+    await response.send(ourStatus)
+    discard await peer.handleStatusV2(peer.networkState, theirStatus)
 
   proc ping(peer: Peer, value: uint64): uint64
     {.libp2pProtocol("ping", 1).} =
@@ -251,10 +281,10 @@ proc setStatusV2Msg(peer: Peer, statusMsg: StatusMsgV2) =
   peer.state(PeerSync).statusMsgV2 = statusMsg
   peer.state(PeerSync).statusLastTime = Moment.now()
 
-proc handleStatus(peer: Peer,
-                  state: PeerSyncNetworkState,
-                  theirStatus: StatusMsg): Future[bool]
-                  {.async: (raises: [CancelledError]).} =
+proc handleStatusV1(peer: Peer,
+                    state: PeerSyncNetworkState,
+                    theirStatus: StatusMsg): Future[bool]
+                    {.async: (raises: [CancelledError]).} =
   let
     res = checkStatusMsg(state, theirStatus)
 
@@ -271,28 +301,77 @@ proc handleStatus(peer: Peer,
       await peer.handlePeer()
     true
 
+proc handleStatusV2(peer: Peer,
+                    state: PeerSyncNetworkState,
+                    theirStatus: StatusMsgV2): Future[bool]
+                    {.async: (raises: [CancelledError]).} =
+  let
+    res = checkStatusMsg(state, theirStatus)
+
+  return if res.isErr():
+    debug "Irrelevant peer", peer, theirStatus, err = res.error()
+    await peer.disconnect(IrrelevantNetwork)
+    false
+  else:
+    peer.setStatusV2Msg(theirStatus)
+
+    if peer.connectionState == Connecting:
+      # As soon as we get here it means that we passed handshake succesfully. So
+      # we can add this peer to PeerPool.
+      await peer.handlePeer()
+    true
+
 proc updateStatus*(peer: Peer): Future[bool] {.async: (raises: [CancelledError]).} =
   ## Request `status` of remote peer ``peer``.
   let
     nstate = peer.networkState(PeerSync)
-    ourStatus = getCurrentStatus(nstate)
-    theirStatus =
-      (await peer.status(ourStatus, timeout = RESP_TIMEOUT_DUR)).valueOr:
-        return false
 
-  await peer.handleStatus(nstate, theirStatus)
+  if nstate.getBeaconTime().slotOrZero.epoch() >= nstate.cfg.FULU_FORK_EPOCH:
+    let
+      ourStatus = getCurrentStatusV2(nstate)
+      theirStatus =
+        (await peer.statusV2(ourStatus, timeout = RESP_TIMEOUT_DUR)).valueOr:
+          return false
+
+    await peer.handleStatusV2(nstate, theirStatus)
+  else:
+    let
+      ourStatus = getCurrentStatusV1(nstate)
+      theirStatus =
+        (await peer.statusV1(ourStatus, timeout = RESP_TIMEOUT_DUR)).valueOr:
+          return false
+
+    await peer.handleStatusV1(nstate, theirStatus)
 
 proc getHeadRoot*(peer: Peer): Eth2Digest =
-  ## Returns head root for specific peer ``peer``.
-  peer.state(PeerSync).statusMsg.headRoot
+  let
+    state = peer.networkState(PeerSync)
+    pstate = peer.state(PeerSync)
+    remoteFork = state.getBeaconTime().slotOrZero.epoch()
+  if remoteFork >= state.cfg.FULU_FORK_EPOCH:
+    pstate.statusMsgV2.headRoot
+  else:
+    pstate.statusMsg.headRoot
 
 proc getHeadSlot*(peer: Peer): Slot =
-  ## Returns head slot for specific peer ``peer``.
-  peer.state(PeerSync).statusMsg.headSlot
+  let
+    state = peer.networkState(PeerSync)
+    pstate = peer.state(PeerSync)
+    remoteFork = state.getBeaconTime().slotOrZero.epoch()
+  if remoteFork >= state.cfg.FULU_FORK_EPOCH:
+    pstate.statusMsgV2.headSlot
+  else:
+    pstate.statusMsg.headSlot
 
 proc getFinalizedEpoch*(peer: Peer): Epoch =
-  ## Returns head slot for specific peer ``peer``.
-  peer.state(PeerSync).statusMsg.finalizedEpoch
+  let
+    state = peer.networkState(PeerSync)
+    pstate = peer.state(PeerSync)
+    remoteFork = state.getBeaconTime().slotOrZero.epoch()
+  if remoteFork >= state.cfg.FULU_FORK_EPOCH:
+    pstate.statusMsgV2.finalizedEpoch
+  else:
+    pstate.statusMsg.finalizedEpoch
 
 proc getStatusLastTime*(peer: Peer): chronos.Moment =
   ## Returns head slot for specific peer ``peer``.

beacon_chain/networking/peer_scores.nim

@@ -35,6 +35,8 @@ const
     ## Peer's answer to our request is fine.
   PeerScoreBadValues* = -1000
     ## Peer's response contains incorrect data.
+  PeerScoreBadColumnIntersection* = -2
+    ## Peer custodies irrelevant custody columns
   PeerScoreBadResponse* = -1000
     ## Peer's response is not in requested range.
   PeerScoreMissingValues* = -25