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,8 @@ 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 +72,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

@@ -24,6 +24,7 @@ const
     ## Enough for finalization in an alternative fork
   MaxBlobless = SLOTS_PER_EPOCH
     ## Arbitrary
+  MaxColumnless = SLOTS_PER_EPOCH
   MaxUnviables = 16 * 1024
     ## About a day of blocks - most likely not needed but it's quite cheap..
 
@@ -57,6 +58,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
@@ -236,6 +239,18 @@ func cleanupBlobless(quarantine: var Quarantine, finalizedSlot: Slot) =
     quarantine.addUnviable k
     quarantine.blobless.del k
 
+func cleanupColumnless(quarantine: var Quarantine, finalizedSlot: Slot) =
+  var toDel: seq[Eth2Digest]
+
+  for k,v in quarantine.columnless:
+    withBlck(v):
+      if not isViable(finalizedSlot, forkyBlck.message.slot):
+        toDel.add k
+
+  for k in toDel:
+    quarantine.addUnviable k
+    quarantine.columnless.del k
+
 func clearAfterReorg*(quarantine: var Quarantine) =
   ## Clear missing and orphans to start with a fresh slate in case of a reorg
   ## Unviables remain unviable and are not cleared.
@@ -336,3 +351,39 @@ func popBlobless*(
 iterator peekBlobless*(quarantine: var Quarantine): ForkedSignedBeaconBlock =
   for k, v in quarantine.blobless.mpairs():
     yield v
+
+proc addColumnless*(
+    quarantine: var Quarantine, finalizedSlot: Slot,
+    signedBlock: deneb.SignedBeaconBlock | electra.SignedBeaconBlock): bool =
+
+  if not isViable(finalizedSlot, signedBlock.message.slot):
+    quarantine.addUnviable(signedBlock.root)
+    return false
+
+  quarantine.cleanupColumnless(finalizedSlot)
+
+  if quarantine.columnless.lenu64 >= MaxColumnless:
+    var oldest_columnless_key: Eth2Digest
+    for k in quarantine.columnless.keys:
+      oldest_columnless_key = k
+      break
+    quarantine.columnless.del oldest_columnless_key
+
+  debug "block quarantine: Adding columnless", blck = shortLog(signedBlock)
+  quarantine.columnless[signedBlock.root] =
+    ForkedSignedBeaconBlock.init(signedBlock)
+  quarantine.missing.del(signedBlock.root)
+  true
+
+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,102 @@
+# 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 data columns in at the same time, so one gets blob insert -> block resolve
+    # -> data columns insert sequence, which leaves garbage data columns.
+    #
+    # This also therefore automatically garbage-collects otherwise valid garbage
+    # data columns 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)
+  r
+
+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): T = T()
+