add: engine_getBlobsV3 helpers for cell level dissemination

beacon_chain/el/el_manager.nim

@@ -631,6 +631,13 @@ proc sendGetBlobsV2toSingleEl(
   let rpcClient = await connection.connectedRpcClient()
   await rpcClient.engine_getBlobsV2(versioned_hashes)
 
+proc sendGetBlobsV3toSingleEl(
+    connection: ELConnection,
+    versioned_hashes: seq[engine_api.VersionedHash]
+): Future[GetBlobsV3Response] {.async: (raises: [CatchableError]).} =
+  let rpcClient = await connection.connectedRpcClient()
+  await rpcClient.engine_getBlobsV3(versioned_hashes)
+
 type
   StatusRelation = enum
     newStatusIsPreferable
@@ -822,6 +829,72 @@ proc sendGetBlobsV2*(
 
     err()
 
+proc sendGetBlobsV3*(
+    m: ElManager,
+    blck: fulu.SignedBeaconBlock
+): Future[Opt[seq[Opt[BlobAndProofV2]]]] {.async: (raises: [CancelledError]).} =
+  if m.elConnections.len == 0:
+    return err()
+
+  when blck is gloas.SignedBeaconBlock:
+    debugGloasComment "handle correctly for Gloas?"
+    return err()
+  else:
+    let deadline = sleepAsync(GETBLOBS_TIMEOUT)
+
+    var bestIdx: Opt[int]
+
+    while true:
+      let requests = m.elConnections.mapIt(
+        sendGetBlobsV3toSingleEl(it,
+          mapIt(blck.message.body.blob_kzg_commitments,
+                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/spec/datatypes/fulu.nim

@@ -20,7 +20,7 @@ import
   "."/[phase0, base, bellatrix, electra],
   chronicles,
   json_serialization,
-  ssz_serialization/[merkleization, proofs],
+  ssz_serialization/[merkleization, proofs, bitseqs],
   ssz_serialization/types as sszTypes,
   ../digest,
   kzg4844/[kzg, kzg_abi]
@@ -110,6 +110,12 @@ type
     block_root*: Eth2Digest
     indices*: DataColumnIndices
 
+  # https://github.com/MarcoPolo/consensus-specs/blob/c02a3a764d9b9cfe74f701493e08aa8291f40dfe/specs/fulu/p2p-interface.md#partial-columns
+  PartialDataColumnSidecar* = object
+    cells_present_bitmap*: BitArray[int(MAX_BLOB_COMMITMENTS_PER_BLOCK)]
+    partial_columns*: List[KzgCell, Limit(MAX_BLOB_COMMITMENTS_PER_BLOCK)]
+    kzg_proofs*: deneb.KzgProofs
+
   # https://github.com/ethereum/consensus-specs/blob/v1.6.0-alpha.0/specs/fulu/das-core.md#matrixentry
   MatrixEntry* = object
     cell*: Cell

beacon_chain/spec/peerdas_helpers.nim

@@ -18,7 +18,7 @@ import
   stew/assign2,
   ./crypto,
   ./[helpers, digest],
-  ./datatypes/fulu
+  ./datatypes/[fulu, deneb]
 
 from std/algorithm import sort
 from std/sequtils import toSeq
@@ -266,6 +266,74 @@ proc assemble_data_column_sidecars*(
 
     sidecars
 
+proc assemble_partial_data_column_sidecars*(
+    signed_beacon_block: fulu.SignedBeaconBlock,
+    blobs: seq[KzgBlob], cell_proofs: seq[Opt[KzgProof]]): seq[fulu.PartialDataColumnSidecar] =
+  ## Returns a seq where element i corresponds to column index i.
+  var sidecars = newSeqOfCap[fulu.PartialDataColumnSidecar](CELLS_PER_EXT_BLOB)
+
+  when signed_beacon_block is gloas.SignedBeaconBlock:
+    debugGloasComment "kzg_commitments removed from beaconblock in gloas"
+    return sidecars
+  else:
+    if blobs.len == 0 or blobs.len > int(MAX_BLOB_COMMITMENTS_PER_BLOCK):
+      return sidecars
+    if cell_proofs.len != blobs.len * CELLS_PER_EXT_BLOB:
+      return sidecars
+
+    var cells = newSeq[CellBytes](blobs.len)
+    for i in 0 ..< blobs.len:
+      cells[i] = computeCells(blobs[i]).get
+
+    for columnIndex in 0..<CELLS_PER_EXT_BLOB:
+      var
+        bitmap: BitArray[int(MAX_BLOB_COMMITMENTS_PER_BLOCK)]
+        partialColumn = newSeqOfCap[KzgCell](blobs.len)
+        partialProofs = newSeqOfCap[KzgProof](blobs.len)
+
+      for rowIndex in 0..<blobs.len:
+        let proofOpt = cell_proofs[rowIndex * CELLS_PER_EXT_BLOB + columnIndex]
+        if proofOpt.isSome:
+          bitmap[Natural(rowIndex)] = true
+          partialColumn.add(cells[rowIndex][columnIndex])
+          partialProofs.add(proofOpt.get)
+
+      sidecars.add fulu.PartialDataColumnSidecar(
+        cells_present_bitmap: bitmap,
+        partial_columns: DataColumn.init(partialColumn),
+        kzg_proofs: deneb.KzgProofs.init(partialProofs))
+
+    sidecars
+
+proc verify_partial_data_column_sidecar_kzg_proofs*(
+    sidecar: fulu.PartialDataColumnSidecar,
+    all_commitments: deneb.KzgCommitments): Result[void, cstring] =
+  ## Verify if the KZG proofs are correct.
+  var
+    cellIndices = newSeqOfCap[CellIndex](sidecar.partial_columns.len)
+    commitments = newSeqOfCap[KzgCommitment](sidecar.partial_columns.len)
+
+  let maxI = min(all_commitments.len, int(MAX_BLOB_COMMITMENTS_PER_BLOCK))
+  for i in 0 ..< maxI:
+    let idx = Natural(i)
+    if sidecar.cells_present_bitmap[idx]:
+      cellIndices.add(CellIndex(i))
+      commitments.add(all_commitments[i])
+
+  if commitments.len != sidecar.partial_columns.len or
+      commitments.len != sidecar.kzg_proofs.len:
+    return err("PartialDataColumnSidecar: length mismatch")
+
+  let res = verifyCellKzgProofBatch(
+      commitments, cellIndices, sidecar.partial_columns.asSeq,
+      sidecar.kzg_proofs.asSeq).valueOr:
+    return err("PartialDataColumnSidecar: validation error")
+
+  if not res:
+    return err("PartialDataColumnSidecar: validation failed")
+
+  ok()
+
 # https://github.com/ethereum/consensus-specs/blob/v1.6.0-beta.1/specs/fulu/p2p-interface.md#verify_data_column_sidecar
 func verify_data_column_sidecar*(cfg: RuntimeConfig, sidecar: fulu.DataColumnSidecar):
                                  Result[void, cstring] =