add getBlobsV2 and payloadV5 (#7123)

* getBlobsV2 and payloadV5

* fix copyright years

* update copyright years

* tiny bug fix

* fix

* refactored reaching ELManager

* removed data column quarantine

* reflect new quarantine changes

* add bpo numbers

* rm oncolumn callback

* fix column quarantine test

* add: bpo parsing (#7195)

* bpo parsing

* fix some formatting stuff

* strip spaces due to indentation

* handle one more case

* stop rejecting blocks with blobs more than electra

* comment out a check

* delay reconstruction from critical path

* check for non zero columns

* make another fix

* ensure BPOs are in correctly sorted order

* bump down parallel requests for data column receiving

* drop put calls to quarantine as we get everything from EL, whenever we get

* remove reconstruction

---------

Co-authored-by: tersec <tersec@users.noreply.github.com>

* use newSeqOfCap

* adding a further comment

---------

Co-authored-by: tersec <tersec@users.noreply.github.com>

Author: agnxsh

beacon_chain/consensus_object_pools/blob_quarantine.nim

@@ -48,6 +48,7 @@ type
 
   BlobQuarantine* =
     SidecarQuarantine[BlobSidecar, OnBlobSidecarCallback]
+  # TODO this OnDataColumnSidecarCallback tpe is notional/unused
   ColumnQuarantine* =
     SidecarQuarantine[DataColumnSidecar, OnDataColumnSidecarCallback]
 
@@ -632,11 +633,6 @@ template onBlobSidecarCallback*(
 ): OnBlobSidecarCallback =
   quarantine.onSidecarCallback
 
-template onDataColumnSidecarCallback*(
-    quarantine: ColumnQuarantine
-): OnDataColumnSidecarCallback =
-  quarantine.onSidecarCallback
-
 func init*(
     T: typedesc[BlobQuarantine],
     cfg: RuntimeConfig,
@@ -660,7 +656,6 @@ func init*(
     T: typedesc[ColumnQuarantine],
     cfg: RuntimeConfig,
     custodyColumns: openArray[ColumnIndex],
-    onDataColumnSidecarCallback: OnDataColumnSidecarCallback
 ): ColumnQuarantine =
   doAssert(len(custodyColumns) <= NUMBER_OF_COLUMNS)
   let size = maxSidecars(NUMBER_OF_COLUMNS)
@@ -679,5 +674,4 @@ func init*(
     indexMap: indexMap,
     custodyColumns: @custodyColumns,
     custodyMap: ColumnMap.init(custodyColumns),
-    onSidecarCallback: onDataColumnSidecarCallback
   )

beacon_chain/el/el_manager.nim

@@ -51,7 +51,9 @@ type
     BellatrixExecutionPayloadWithValue |
     GetPayloadV2Response |
     GetPayloadV3Response |
-    GetPayloadV4Response
+    GetPayloadV4Response |
+    GetPayloadV5Response
+
 
 const
   noTimeout = WithoutTimeout(0)
@@ -64,6 +66,8 @@ 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
+
   connectionStateChangeHysteresisThreshold = 15
     ## How many unsuccesful/successful requests we must see
     ## before declaring the connection as degraded/restored
@@ -406,9 +410,11 @@ proc getPayloadFromSingleEL(
             suggestedFeeRecipient: suggestedFeeRecipient,
             withdrawals: withdrawals))
       elif  GetPayloadResponseType is engine_api.GetPayloadV3Response or
-            GetPayloadResponseType is engine_api.GetPayloadV4Response:
+            GetPayloadResponseType is engine_api.GetPayloadV4Response or
+            GetPayloadResponseType is engine_api.GetPayloadV5Response:
         # https://github.com/ethereum/execution-apis/blob/90a46e9137c89d58e818e62fa33a0347bba50085/src/engine/prague.md
         # does not define any new forkchoiceUpdated, so reuse V3 from Dencun
+        # https://github.com/ethereum/execution-apis/blob/5d634063ccfd897a6974ea589c00e2c1d889abc9/src/engine/osaka.md
         let response = await rpcClient.forkchoiceUpdated(
           ForkchoiceStateV1(
             headBlockHash: headBlock.asBlockHash,
@@ -459,7 +465,7 @@ template EngineApiResponseType*(T: type electra.ExecutionPayloadForSigning): typ
   engine_api.GetPayloadV4Response
 
 template EngineApiResponseType*(T: type fulu.ExecutionPayloadForSigning): type =
-  engine_api.GetPayloadV4Response
+  engine_api.GetPayloadV5Response
 
 template toEngineWithdrawals*(withdrawals: seq[capella.Withdrawal]): seq[WithdrawalV1] =
   mapIt(withdrawals, toEngineWithdrawal(it))
@@ -631,6 +637,13 @@ proc sendNewPayloadToSingleEL(
     payload, versioned_hashes, Hash32 parent_beacon_block_root,
     executionRequests)
 
+proc sendGetBlobsV2toSingleEl(
+    connection: ELConnection,
+    versioned_hashes: seq[engine_api.VersionedHash]
+): Future[GetBlobsV2Response] {.async: (raises: [CatchableError]).} =
+  let rpcClient = await connection.connectedRpcClient()
+  await rpcClient.engine_getBlobsV2(versioned_hashes)
+
 type
   StatusRelation = enum
     newStatusIsPreferable
@@ -756,6 +769,69 @@ proc lazyWait(futures: seq[FutureBase]) {.async: (raises: []).} =
     if len(pending) > 0:
       await noCancel allFutures(pending)
 
+proc sendGetBlobsV2*(
+    m: ELManager,
+    blck: fulu.SignedBeaconBlock,
+): Future[Opt[seq[BlobAndProofV2]]] {.async: (raises: [CancelledError]).} =
+
+  if m.elConnections.len == 0:
+    return err()
+
+  let deadline = sleepAsync(GETBLOBS_TIMEOUT)
+
+  var bestIdx: Opt[int] = Opt.none(int)
+
+  while true:
+    let requests = m.elConnections.mapIt(
+      sendGetBlobsV2toSingleEl(it,
+        mapIt(blck.message.body.blob_kzg_commitments,
+              engine_api.VersionedHash(kzg_commitment_to_versioned_hash(it)))
+      )
+    )
+
+    let timeoutExceeded =
+      try:
+        await allFutures(requests).wait(deadline)
+        false
+      except AsyncTimeoutError:
+        true
+      except CancelledError as exc:
+        # cancel anything still running, then re-raise
+        await noCancel allFutures(
+          requests.filterIt(not it.finished()).mapIt(it.cancelAndWait())
+        )
+        raise exc
+
+    for idx, req in requests:
+      if req.finished():
+        # choose the first successful (not failed) response
+        if req.error.isNil and bestIdx.isNone:
+          bestIdx = Opt.some(idx)
+      else:
+        # finished == false
+        let errmsg =
+          if req.error.isNil: "request still pending"
+          else: req.error.msg
+        warn "Timeout while getting blobs & proofs",
+             url = m.elConnections[idx].engineUrl.url,
+             reason = errmsg
+
+    await noCancel allFutures(
+      requests.filterIt(not it.finished()).mapIt(it.cancelAndWait())
+    )
+
+    if bestIdx.isSome():
+      let chosen = requests[bestIdx.get()]
+      # chosen is finished; but could still be an error, so guard again
+      if chosen.error.isNil:
+        return ok(chosen.value())
+      else:
+        warn "Chosen EL failed unexpectedly", reason = chosen.error.msg
+    if timeoutExceeded:
+      break
+
+  err()
+
 proc sendNewPayload*(
     m: ELManager,
     blck: SomeForkyBeaconBlock,

beacon_chain/el/engine_api_conversions.nim

@@ -1,5 +1,5 @@
 # beacon_chain
-# Copyright (c) 2024 Status Research & Development GmbH
+# Copyright (c) 2024-2025 Status Research & Development GmbH
 # Licensed and distributed under either of
 #   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
 #   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
@@ -224,7 +224,7 @@ func asConsensusType*(
     executionRequests: payload.executionRequests)
 
 func asConsensusTypeFulu*(
-    payload: GetPayloadV4Response):
+    payload: GetPayloadV5Response):
     fulu.ExecutionPayloadForSigning =
   fulu.ExecutionPayloadForSigning(
     executionPayload: payload.executionPayload.asFuluConsensusPayload,

beacon_chain/gossip_processing/block_processor.nim

@@ -283,7 +283,8 @@ proc storeBackfillBlock(
     self.consensusManager.dag.db.putBlobSidecar(b[])
 
   # Only store data columns after successfully establishing block validity
-  let columns = dataColumnsOpt.valueOr: DataColumnSidecars @[]
+  let
+    columns = dataColumnsOpt.valueOr: DataColumnSidecars @[]
   for c in columns:
     self.consensusManager.dag.db.putDataColumnSidecar(c[])
 
@@ -728,16 +729,17 @@ proc storeBlock(
 
   when typeof(signedBlock).kind >= ConsensusFork.Fulu:
     if dataColumnsOpt.isSome:
-      let columns = dataColumnsOpt.get()
-      let kzgCommits = signedBlock.message.body.blob_kzg_commitments.asSeq
-      if columns.len > 0 and kzgCommits.len > 0:
-        for i in 0..<columns.len:
+      let
+        columns0 = dataColumnsOpt.get()
+        kzgCommits = signedBlock.message.body.blob_kzg_commitments.asSeq
+      if columns0.len > 0 and kzgCommits.len > 0:
+        for i in 0..<columns0.len:
           let r =
-            verify_data_column_sidecar_kzg_proofs(columns[i][])
+            verify_data_column_sidecar_kzg_proofs(columns0[i][])
           if r.isErr:
             debug "data column validation failed",
               blockRoot = shortLog(signedBlock.root),
-              column_sidecar = shortLog(columns[i][]),
+              column_sidecar = shortLog(columns0[i][]),
               blck = shortLog(signedBlock.message),
               signature = shortLog(signedBlock.signature),
               msg = r.error()

beacon_chain/gossip_processing/eth2_processor.nim

@@ -8,9 +8,12 @@
 {.push raises: [].}
 
 import
-  std/[tables],
+  std/[tables, sequtils],
   chronicles, chronos, metrics,
   taskpools,
+  kzg4844/kzg,
+  ssz_serialization/types,
+  ../el/el_manager,
   ../spec/[helpers, forks, peerdas_helpers],
   ../consensus_object_pools/[
     blob_quarantine, block_clearance, block_quarantine, blockchain_dag,
@@ -349,15 +352,80 @@ proc processBlobSidecar*(
 
   v
 
+proc validateDataColumnSidecarFromEL*(
+    self: ref Eth2Processor,
+    block_root: Eth2Digest):
+    Future[ValidationRes]
+    {.async: (raises: [CancelledError]).} =
+  let elManager = self.blockProcessor[].consensusManager.elManager
+  if (let o = self.quarantine[].popColumnless(block_root); o.isSome):
+    let columnless = o.unsafeGet()
+    withBlck(columnless):
+      when consensusFork >= ConsensusFork.Fulu:
+        let
+          start_time = Moment.now()
+        let blobsFromElOpt =
+          await elManager.sendGetBlobsV2(forkyBlck)
+        if blobsFromElOpt.isSome():
+          let blobsEl = blobsFromElOpt.get()
+
+          # check lengths of array[BlobAndProofV2 with blobs
+          # kzg commitments of the signed block
+          if blobsEl.len == forkyBlck.message.body.blob_kzg_commitments.len:
+
+            # we have received all columns from the EL
+            # hence we can safely remove the columnless block from quarantine
+            var flat_proof: seq[kzg.KzgProof] = @[]
+            for item in blobsEl:
+              for proof in item.proofs:
+                flat_proof.add(kzg.KzgProof(bytes: proof.data))
+
+            let
+              recovered_columns =
+                assemble_data_column_sidecars(
+                  forkyBlck,
+                  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")
+
 proc processDataColumnSidecar*(
-    self: var Eth2Processor, src: MsgSource,
-    dataColumnSidecar: DataColumnSidecar, subnet_id: uint64): ValidationRes =
+    self: ref Eth2Processor, src: MsgSource,
+    dataColumnSidecar: DataColumnSidecar, subnet_id: uint64):
+    Future[ValidationRes] {.async: (raises: [CancelledError]).} =
   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
 
   let
     wallTime = self.getCurrentBeaconTime()
     (_, wallSlot) = wallTime.toSlot()
-    block_root = hash_tree_root(block_header)
 
   logScope:
     dcs = shortLog(dataColumnSidecar)
@@ -387,17 +455,10 @@ proc processDataColumnSidecar*(
         let cres =
           self.dataColumnQuarantine[].popSidecars(block_root, forkyBlck)
         if cres.isSome():
-          if cres.get().lenu64 > (self.dag.cfg.NUMBER_OF_COLUMNS div 2):
-            # We have enough data columns to reconstruct the rest
-            let
-              recovered_cps = recover_cells_and_proofs(cres.get())
-              reconstructed_columns =
-                reconstruct_data_column_sidecars(forkyBlck, recovered_cps.get)
-
-            self.blockProcessor[].enqueueBlock(
-              MsgSource.gossip, columnless,
-              Opt.none(BlobSidecars),
-              Opt.some(reconstructed_columns))
+          self.blockProcessor[].enqueueBlock(
+            MsgSource.gossip, columnless,
+            Opt.none(BlobSidecars),
+            cres)
         else:
           discard self.quarantine[].addColumnless(
             self.dag.finalizedHead.slot, forkyBlck)

beacon_chain/gossip_processing/gossip_validation.nim

@@ -294,7 +294,16 @@ template checkedReject(
   pool.dag.checkedReject(error)
 
 func getMaxBlobsPerBlock(cfg: RuntimeConfig, slot: Slot): uint64 =
-  if slot >= cfg.ELECTRA_FORK_EPOCH.start_slot:
+  if slot >= cfg.FULU_FORK_EPOCH.start_slot:
+    let
+      epoch: Epoch = Epoch(uint64(slot) div SLOTS_PER_EPOCH)
+      maxBlobs = get_max_blobs_per_block_bpo(cfg, epoch)
+    if maxBlobs.isSome():
+      maxBlobs.get()
+    else:
+      # If the max blobs per block is not set, use the default value
+      cfg.MAX_BLOBS_PER_BLOCK_ELECTRA
+  elif slot >= cfg.ELECTRA_FORK_EPOCH.start_slot:
     cfg.MAX_BLOBS_PER_BLOCK_ELECTRA
   else:
     cfg.MAX_BLOBS_PER_BLOCK
@@ -696,12 +705,6 @@ proc validateDataColumnSidecar*(
     if r.isErr:
       return dag.checkedReject(r.error)
 
-  # Send notification about new data column sidecar via callback
-  let onDataColumnSidecarCallback =
-    dataColumnQuarantine[].onDataColumnSidecarCallback()
-  if not(isNil(onDataColumnSidecarCallback)):
-    onDataColumnSidecarCallback data_column_sidecar
-
   ok()
 
 # https://github.com/ethereum/consensus-specs/blob/v1.3.0/specs/phase0/p2p-interface.md#beacon_block
@@ -843,7 +846,7 @@ proc validateBeaconBlock*(
   # validation.
   validateBeaconBlockBellatrix(signed_beacon_block, parent)
 
-  dag.validateBeaconBlockDeneb(signed_beacon_block, wallTime)
+  # dag.validateBeaconBlockDeneb(signed_beacon_block, wallTime)
 
   # [REJECT] The block is from a higher slot than its parent.
   if not (signed_beacon_block.message.slot > parent.bid.slot):