EIP7594: p2p-interface

AllTests-mainnet.md

@@ -55,14 +55,15 @@ OK: 4/4 Fail: 0/4 Skip: 0/4
 + sanity check Deneb states [Preset: mainnet]                                                OK
 + sanity check Deneb states, reusing buffers [Preset: mainnet]                               OK
 + sanity check blobs [Preset: mainnet]                                                       OK
++ sanity check data columns [Preset: mainnet]                                                OK
 + sanity check genesis roundtrip [Preset: mainnet]                                           OK
 + sanity check phase 0 blocks [Preset: mainnet]                                              OK
 + sanity check phase 0 getState rollback [Preset: mainnet]                                   OK
 + sanity check phase 0 states [Preset: mainnet]                                              OK
 + sanity check phase 0 states, reusing buffers [Preset: mainnet]                             OK
 + sanity check state diff roundtrip [Preset: mainnet]                                        OK
 ```
-OK: 25/25 Fail: 0/25 Skip: 0/25
+OK: 26/26 Fail: 0/26 Skip: 0/26
 ## Beacon state [Preset: mainnet]
 ```diff
 + Smoke test initialize_beacon_state_from_eth1 [Preset: mainnet]                             OK
@@ -1154,4 +1155,4 @@ OK: 2/2 Fail: 0/2 Skip: 0/2
 OK: 9/9 Fail: 0/9 Skip: 0/9
 
 ---TOTAL---
-OK: 803/808 Fail: 0/808 Skip: 5/808
+OK: 804/809 Fail: 0/809 Skip: 5/809

beacon_chain/beacon_chain_db.nim

@@ -254,6 +254,13 @@ func blobkey(root: Eth2Digest, index: BlobIndex) : array[40, byte] =
 
   ret
 
+func columnkey(root: Eth2Digest, index: ColumnIndex) : array[40, byte] =
+  var ret: array[40, byte]
+  ret[0..<8] = toBytes(index)
+  ret[8..<40] = root.data
+
+  ret
+
 template expectDb(x: auto): untyped =
   # There's no meaningful error handling implemented for a corrupt database or
   # full disk - this requires manual intervention, so we'll panic for now
@@ -808,11 +815,22 @@ proc putBlobSidecar*(
   let block_root = hash_tree_root(value.signed_block_header.message)
   db.blobs.putSZSSZ(blobkey(block_root, value.index), value)
 
+proc putDataColumnSidecar*(
+    db: BeaconChainDB,
+    value: DataColumnSidecar) =
+  let block_root = hash_tree_root(value.signed_block_header.message)
+  db.blobs.putSZSSZ(columnkey(block_root, value.index), value)
+
 proc delBlobSidecar*(
     db: BeaconChainDB,
     root: Eth2Digest, index: BlobIndex): bool =
   db.blobs.del(blobkey(root, index)).expectDb()
 
+proc delDataColumnSidecar*(
+    db: BeaconChainDB,
+    root: Eth2Digest, index: ColumnIndex): bool =
+  db.blobs.del(columnkey(root, index)).expectDb()
+
 proc updateImmutableValidators*(
     db: BeaconChainDB, validators: openArray[Validator]) =
   # Must be called before storing a state that references the new validators
@@ -1071,6 +1089,17 @@ proc getBlobSidecar*(db: BeaconChainDB, root: Eth2Digest, index: BlobIndex,
                      value: var BlobSidecar): bool =
   db.blobs.getSZSSZ(blobkey(root, index), value) == GetResult.found
 
+proc getDataColumnSidecarSZ*(db: BeaconChainDB, root: Eth2Digest, 
+                             index: ColumnIndex, data: var seq[byte]): bool =
+  let dataPtr = addr data # Short-lived
+  func decode(data: openArray[byte]) =
+    assign(dataPtr[], data)
+  db.blobs.get(columnkey(root, index), decode).expectDb()
+
+proc getDataColumnSidecar*(db: BeaconChainDB, root: Eth2Digest, index: ColumnIndex,
+                           value: var DataColumnSidecar): bool =
+  db.blobs.getSZSSZ(columnkey(root, index), value) == GetResult.found
+
 proc getBlockSZ*(
     db: BeaconChainDB, key: Eth2Digest, data: var seq[byte],
     T: type phase0.TrustedSignedBeaconBlock): bool =

beacon_chain/beacon_node.nim

@@ -22,7 +22,7 @@ import
   ./el/el_manager,
   ./consensus_object_pools/[
     blockchain_dag, blob_quarantine, block_quarantine, consensus_manager,
-    attestation_pool, sync_committee_msg_pool, validator_change_pool],
+    data_column_quarantine, attestation_pool, sync_committee_msg_pool, validator_change_pool],
   ./spec/datatypes/[base, altair],
   ./spec/eth2_apis/dynamic_fee_recipients,
   ./sync/[sync_manager, request_manager],
@@ -71,6 +71,7 @@ type
     dag*: ChainDAGRef
     quarantine*: ref Quarantine
     blobQuarantine*: ref BlobQuarantine
+    dataColumnQuarantine*: ref DataColumnQuarantine
     attestationPool*: ref AttestationPool
     syncCommitteeMsgPool*: ref SyncCommitteeMsgPool
     lightClientPool*: ref LightClientPool

beacon_chain/consensus_object_pools/block_quarantine.nim

@@ -57,6 +57,8 @@ type
       ## all blobs for this block, we can proceed to resolving the
       ## block as well. A blobless block inserted into this table must
       ## have a resolved parent (i.e., it is not an orphan).
+
+    columnless: OrderedTable[Eth2Digest, ForkedSignedBeaconBlock]
 
     unviable*: OrderedTable[Eth2Digest, tuple[]]
       ## Unviable blocks are those that come from a history that does not
@@ -336,3 +338,16 @@ func popBlobless*(
 iterator peekBlobless*(quarantine: var Quarantine): ForkedSignedBeaconBlock =
   for k, v in quarantine.blobless.mpairs():
     yield v
+
+func popColumnless*(
+    quarantine: var Quarantine,
+    root: Eth2Digest): Opt[ForkedSignedBeaconBlock] =
+  var blck: ForkedSignedBeaconBlock
+  if quarantine.columnless.pop(root, blck):
+    Opt.some(blck)
+  else:
+    Opt.none(ForkedSignedBeaconBlock)
+
+iterator peekColumnless*(quarantine: var Quarantine): ForkedSignedBeaconBlock =
+  for k,v in quarantine.columnless.mpairs():
+    yield v

beacon_chain/consensus_object_pools/data_column_quarantine.nim

@@ -0,0 +1,103 @@
+# beacon_chain
+# Copyright (c) 2018-2024 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).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+{.push raises: [].}
+
+import
+  std/tables,
+  ../spec/[helpers, eip7594_helpers]
+
+from std/sequtils import mapIt
+from std/strutils import join
+
+const
+  MaxDataColumns = 3 * SLOTS_PER_EPOCH * NUMBER_OF_COLUMNS
+  ## Same limit as `MaxOrphans` in `block_quarantine`
+  ## data columns may arrive before an orphan is tagged `columnless`
+
+type
+  DataColumnQuarantine* = object
+    data_columns*:
+      OrderedTable[(Eth2Digest, ColumnIndex), ref DataColumnSidecar]
+    onDataColumnSidecarCallback*: OnDataColumnSidecarCallback
+
+  DataColumnFetchRecord* = object
+    block_root*: Eth2Digest
+    indices*: seq[ColumnIndex]
+
+  OnDataColumnSidecarCallback = proc(data: DataColumnSidecar) {.gcsafe, raises: [].}
+
+func shortLog*(x: seq[ColumnIndex]): string =
+  "<" & x.mapIt($it).join(", ") & ">"
+
+func shortLog*(x: seq[DataColumnFetchRecord]): string =
+  "[" & x.mapIt(shortLog(it.block_root) & shortLog(it.indices)).join(", ") & "]"
+
+func put*(quarantine: var DataColumnQuarantine, dataColumnSidecar: ref DataColumnSidecar) =
+  if quarantine.data_columns.lenu64 >= MaxDataColumns:
+    # FIFO if full. For example, sync manager and request manager can race to
+    # put blobs in at the same time, so one gets blob insert -> block resolve
+    # -> blob insert sequence, which leaves garbage blobs.
+    #
+    # This also therefore automatically garbage-collects otherwise valid garbage
+    # blobs which are correctly signed, point to either correct block roots or a
+    # block root which isn't ever seen, and then are for any reason simply never
+    # used.
+    var oldest_column_key: (Eth2Digest, ColumnIndex)
+    for k in quarantine.data_columns.keys:
+      oldest_column_key = k
+      break
+    quarantine.data_columns.del(oldest_column_key)
+  let block_root = hash_tree_root(dataColumnSidecar.signed_block_header.message)
+  discard quarantine.data_columns.hasKeyOrPut(
+    (block_root, dataColumnSidecar.index), dataColumnSidecar)
+
+func hasDataColumn*(
+    quarantine: DataColumnQuarantine,
+    slot: Slot,
+    proposer_index: uint64,
+    index: ColumnIndex): bool =
+  for data_column_sidecar in quarantine.data_columns.values:
+    template block_header: untyped = data_column_sidecar.signed_block_header.message
+    if block_header.slot == slot and
+        block_header.proposer_index == proposer_index and
+        data_column_sidecar.index == index:
+      return true
+  false
+
+func popDataColumns*(
+    quarantine: var DataColumnQuarantine, digest: Eth2Digest,
+    blck: deneb.SignedBeaconBlock | electra.SignedBeaconBlock):
+    seq[ref DataColumnSidecar] =
+  var r: seq[ref DataColumnSidecar]
+  for idx in 0..<len(blck.message.body.blob_kzg_commitments):
+    var c: ref DataColumnSidecar
+    if quarantine.data_columns.pop((digest, ColumnIndex idx), c):
+      r.add(c)
+  true
+
+func hasDataColumns*(quarantine: DataColumnQuarantine,
+    blck: deneb.SignedBeaconBlock | electra.SignedBeaconBlock): bool =
+  for idx in 0..<len(blck.message.body.blob_kzg_commitments):
+    if (blck.root, ColumnIndex idx) notin quarantine.data_columns:
+      return false
+  true
+
+func dataColumnFetchRecord*(quarantine: DataColumnQuarantine,
+    blck: deneb.SignedBeaconBlock | electra.SignedBeaconBlock): DataColumnFetchRecord =
+  var indices: seq[ColumnIndex]
+  for i in 0..<len(blck.message.body.blob_kzg_commitments):
+    let idx = ColumnIndex(i)
+    if not quarantine.data_columns.hasKey(
+        (blck.root, idx)):
+      indices.add(idx)
+  DataColumnFetchRecord(block_root: blck.root, indices: indices)
+
+func init*(
+    T: type DataColumnQuarantine, onDataColumnSidecarCallback: OnDataColumnSidecarCallback): T =
+  T(onDataColumnSidecarCallback: onDataColumnSidecarCallback)
+

beacon_chain/gossip_processing/eth2_processor.nim

@@ -15,8 +15,8 @@ import
   ../spec/datatypes/[altair, phase0, deneb, eip7594],
   ../consensus_object_pools/[
     blob_quarantine, block_clearance, block_quarantine, blockchain_dag,
-    attestation_pool, light_client_pool, sync_committee_msg_pool,
-    validator_change_pool],
+    data_column_quarantine, attestation_pool, light_client_pool, 
+    sync_committee_msg_pool, validator_change_pool],
   ../validators/validator_pool,
   ../beacon_clock,
   "."/[gossip_validation, block_processor, batch_validation],
@@ -156,6 +156,8 @@ type
 
     blobQuarantine*: ref BlobQuarantine
 
+    dataColumnQuarantine*: ref DataColumnQuarantine
+
     # Application-provided current time provider (to facilitate testing)
     getCurrentBeaconTime*: GetBeaconTimeFn
 
@@ -345,7 +347,7 @@ proc processDataColumnSidecar*(
   debug "Data column received", delay
 
   let v =
-    self.dag.validateDataColumnSidecar(self.quarantine, self.blobQuarantine,
+    self.dag.validateDataColumnSidecar(self.quarantine, self.dataColumnQuarantine,
                                  dataColumnSidecar, wallTime, subnet_id)
 
   if v.isErr():

beacon_chain/gossip_processing/gossip_validation.nim

@@ -16,7 +16,7 @@ import
     beaconstate, state_transition_block, forks, helpers, network, signatures, eip7594_helpers],
   ../consensus_object_pools/[
     attestation_pool, blockchain_dag, blob_quarantine, block_quarantine,
-    spec_cache, light_client_pool, sync_committee_msg_pool,
+    data_column_quarantine, spec_cache, light_client_pool, sync_committee_msg_pool,
     validator_change_pool],
   ".."/[beacon_clock],
   ./batch_validation
@@ -490,7 +490,7 @@ proc validateBlobSidecar*(
 # https://github.com/ethereum/consensus-specs/blob/5f48840f4d768bf0e0a8156a3ed06ec333589007/specs/_features/eip7594/p2p-interface.md#the-gossip-domain-gossipsub
 proc validateDataColumnSidecar*(
     dag: ChainDAGRef, quarantine: ref Quarantine,
-    blobQuarantine: ref BlobQuarantine, data_column_sidecar: DataColumnSidecar,
+    dataColumnQuarantine: ref DataColumnQuarantine, data_column_sidecar: DataColumnSidecar,
     wallTime: BeaconTime, subnet_id: uint64): Result[void, ValidationError] =
 
   template block_header: untyped = data_column_sidecar.signed_block_header.message
@@ -538,7 +538,7 @@ proc validateDataColumnSidecar*(
   let block_root = hash_tree_root(block_header)
   if dag.getBlockRef(block_root).isSome():
     return errIgnore("DataColumnSidecar: already have block")
-  if blobQuarantine[].hasBlob(
+  if dataColumnQuarantine[].hasDataColumn(
       block_header.slot, block_header.proposer_index, data_column_sidecar.index):
     return errIgnore("DataColumnSidecar: already have valid data column from same proposer")
 

beacon_chain/nimbus_beacon_node.nim

@@ -14,6 +14,7 @@ import
   stew/[byteutils, io2],
   eth/p2p/discoveryv5/[enr, random2],
   ./consensus_object_pools/blob_quarantine,
+  ./consensus_object_pools/data_column_quarantine,
   ./consensus_object_pools/vanity_logs/vanity_logs,
   ./networking/[topic_params, network_metadata_downloads],
   ./rpc/[rest_api, state_ttl_cache],
@@ -1410,6 +1411,24 @@ proc pruneBlobs(node: BeaconNode, slot: Slot) =
               count = count + 1
     debug "pruned blobs", count, blobPruneEpoch
 
+proc pruneDataColumns(node: BeaconNode, slot: Slot) =
+  let dataColumnPruneEpoch = (slot.epoch -
+                              node.dag.cfg.MIN_EPOCHS_FOR_DATA_COLUMN_SIDECARS_REQUESTS - 1)
+  if slot.is_epoch() and dataColumnPruneEpoch >= node.dag.cfg.DENEB_FORK_EPOCH:
+    var blocks: array[SLOTS_PER_EPOCH.int, BlockId]
+    var count = 0
+    let startIndex = node.dag.getBlockRange(
+      dataColumnPruneEpoch.start_slot, 1, blocks.toopenArray(0, SLOTS_PER_EPOCH - 1))
+    for i in startIndex..<SLOTS_PER_EPOCH:
+      let blck = node.dag.getForkedBlock(blocks[int(i)]).valueOr: continue
+      withBlck(blck):
+        when typeof(forkyBlck).kind < ConsensusFork.Deneb: continue
+        else:
+          for j in 0..len(forkyBlck.message.body.blob_kzg_commitments) - 1:
+            if node.db.delDataColumnSidecar(blocks[int(i)].root, ColumnIndex(j)):
+              count = count + 1
+    debug "pruned data columns", count, dataColumnPruneEpoch
+
 proc onSlotEnd(node: BeaconNode, slot: Slot) {.async.} =
   # Things we do when slot processing has ended and we're about to wait for the
   # next slot
@@ -1444,6 +1463,7 @@ proc onSlotEnd(node: BeaconNode, slot: Slot) {.async.} =
       # the pruning for later
       node.dag.pruneHistory()
       node.pruneBlobs(slot)
+      node.pruneDataColumns(slot)
 
   when declared(GC_fullCollect):
     # The slots in the beacon node work as frames in a game: we want to make

beacon_chain/rpc/rest_constants.nim

@@ -213,6 +213,8 @@ const
     "LC finality update unavailable"
   LCOptUpdateUnavailable* =
     "LC optimistic update unavailable"
+  DataColumnsOutOfRange* =
+    "Requested slot is out of data column window"
   DeprecatedRemovalBeaconBlocksDebugStateV1* =
     "v1/beacon/blocks/{block_id} and v1/debug/beacon/states/{state_id} " &
       "endpoints were deprecated and replaced by v2: " &

beacon_chain/spec/datatypes/eip7594.nim

@@ -9,6 +9,9 @@
 
 import "."/[base, deneb], kzg4844
 
+from std/sequtils import mapIt
+from std/strutils import join
+
 export base
 
 const
@@ -51,8 +54,8 @@ type
   DataColumnSidecar* = object
     index*: ColumnIndex # Index of column in extended matrix
     column*: DataColumn
-    kzg_commitments*: List[KzgCommitment, Limit(MAX_BLOB_COMMITMENTS_PER_BLOCK)]
-    kzg_proofs*: List[KzgProof, Limit(MAX_BLOB_COMMITMENTS_PER_BLOCK)]
+    kzg_commitments*: KzgCommitments
+    kzg_proofs*: KzgProofs
     signed_block_header*: SignedBeaconBlockHeader
     kzg_commitments_inclusion_proof*:
       array[KZG_COMMITMENTS_INCLUSION_PROOF_DEPTH, Eth2Digest]
@@ -74,4 +77,7 @@ func shortLog*(v: DataColumnSidecar): auto =
     kzg_commitments: v.kzg_commitments.len,
     kzg_proofs: v.kzg_proofs.len,
     block_header: shortLog(v.signed_block_header.message),
-  )
+  )
+
+func shortLog*(x: seq[DataColumnIdentifier]): string =
+  "[" & x.mapIt(shortLog(it.block_root) & "/" & $it.index).join(", ") & "]"

beacon_chain/spec/eth2_apis/eth2_rest_serialization.nim

@@ -53,6 +53,7 @@ RestJson.useDefaultSerializationFor(
   Checkpoint,
   Consolidation,
   ContributionAndProof,
+  DataColumnSidecar,
   DataEnclosedObject,
   DataMetaEnclosedObject,
   DataOptimisticAndFinalizedObject,

beacon_chain/spec/network.nim

@@ -34,6 +34,10 @@ const
   MAX_REQUEST_BLOB_SIDECARS*: uint64 =
     MAX_REQUEST_BLOCKS_DENEB * MAX_BLOBS_PER_BLOCK
 
+  # https://github.com/ethereum/consensus-specs/blob/v1.5.0-alpha.2/specs/_features/eip7594/p2p-interface.md#configuration
+  MAX_REQUEST_DATA_COLUMNS*: uint64 =
+    MAX_REQUEST_BLOCKS_DENEB * NUMBER_OF_COLUMNS
+
   defaultEth2TcpPort* = 9000
   defaultEth2TcpPortDesc* = $defaultEth2TcpPort