add gossip validation for dc, and data column quarantine strategy (#6581)

* add gossip validation for dc

* review 1

* rm callback

* review 2

* added custody columns as a global entity

* alpha 8

* few typosA

Author: agnxsh

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],
@@ -73,6 +73,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/data_column_quarantine.nim

@@ -0,0 +1,190 @@
+# 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/datatypes/eip7594,
+  ../spec/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[DataColumnIdentifier, ref DataColumnSidecar]
+    supernode*: bool
+    custody_columns*: seq[ColumnIndex]
+    onDataColumnSidecarCallback*: OnDataColumnSidecarCallback
+  
+  DataColumnFetchRecord* = object
+    block_root*: Eth2Digest
+    indices*: seq[ColumnIndex]
+
+  OnDataColumnSidecarCallback = proc(data: DataColumnSidecar) {.gcsafe, raises: [].}
+
+func init*(T: type DataColumnQuarantine): T =
+  T()
+
+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.len >= static(MaxDataColumns.int):
+    # FIFO if full. For example, sync manager and request manager can race
+    # to put data columns in at the same time, so one gets data column
+    # insert -> block resolve -> data column insert, which leaves
+    # garbage data columns.
+    #
+    # This also therefore automatically garbage-collects otherwise valid 
+    # data columns that are correctly signed, point to either correct block
+    # root which isn't ever seen, and then for any reason simply never used.
+    var oldest_column_key: DataColumnIdentifier
+    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(
+    DataColumnIdentifier(block_root: block_root,
+                         index: 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 peekColumnIndices*(quarantine: DataColumnQuarantine,
+                        blck: electra.SignedBeaconBlock):
+                        seq[ColumnIndex] =
+  # Peeks into the currently received column indices
+  # from quarantine, necessary data availability checks
+  var indices: seq[ColumnIndex]
+  for col_idx in quarantine.custody_columns:
+    if quarantine.data_columns.hasKey(
+        DataColumnIdentifier(block_root: blck.root,
+                             index: ColumnIndex col_idx)):
+      indices.add(col_idx)
+  indices
+
+func gatherDataColumns*(quarantine: DataColumnQuarantine,
+                       digest: Eth2Digest): 
+                       seq[ref DataColumnSidecar] =
+  # Returns the current data columns quried by a 
+  # block header
+  var columns: seq[ref DataColumnSidecar]
+  for i in quarantine.custody_columns:
+    let dc_identifier = 
+      DataColumnIdentifier(
+        block_root: digest,
+        index: i)
+    if quarantine.data_columns.hasKey(dc_identifier):
+      let value = 
+        quarantine.data_columns.getOrDefault(dc_identifier,
+                                             default(ref DataColumnSidecar))
+      columns.add(value)
+  columns
+
+func popDataColumns*(
+    quarantine: var DataColumnQuarantine, digest: Eth2Digest,
+    blck: electra.SignedBeaconBlock):
+    seq[ref DataColumnSidecar] =
+  var r: DataColumnSidecars
+  for idx in quarantine.custody_columns:
+    var c: ref DataColumnSidecar
+    if quarantine.data_columns.pop(
+        DataColumnIdentifier(block_root: digest,
+                             index: idx),
+                             c):
+      r.add(c)
+  r
+
+func hasMissingDataColumns*(quarantine: DataColumnQuarantine,
+    blck: electra.SignedBeaconBlock): bool =
+  # `hasMissingDataColumns` consists of the data columns that,
+  # have been missed over gossip, also in case of a supernode,
+  # the method would return missing columns when the supernode
+  # has not received data columns upto the requisite limit (i.e 50%
+  # of NUMBER_OF_COLUMNS).
+
+  # This method shall be actively used by the `RequestManager` to
+  # root request columns over RPC.
+  var col_counter = 0
+  for idx in quarantine.custody_columns:
+    let dc_identifier = 
+      DataColumnIdentifier(
+        block_root: blck.root,
+        index: idx)
+    if dc_identifier notin quarantine.data_columns:
+      inc col_counter
+  if quarantine.supernode and col_counter != NUMBER_OF_COLUMNS:
+    return false
+  elif quarantine.supernode == false and
+      col_counter != max(SAMPLES_PER_SLOT, CUSTODY_REQUIREMENT):
+    return false
+  else:
+    return true
+
+func hasEnoughDataColumns*(quarantine: DataColumnQuarantine,
+    blck: electra.SignedBeaconBlock): bool =
+  # `hasEnoughDataColumns` dictates whether there is `enough`
+  # data columns for a block to be enqueued, ideally for a supernode
+  # if it receives atleast 50%+ gossip and RPC
+
+  # Once 50%+ columns are available we can use this function to
+  # check it, and thereby check column reconstructability, right from 
+  # gossip validation, consequently populating the quarantine with
+  # rest of the data columns.
+  if quarantine.supernode:
+    let
+      collectedColumns = quarantine.gatherDataColumns(blck.root)
+    if collectedColumns.len >= (quarantine.custody_columns.len div 2):
+      return true
+  else:
+    for i in quarantine.custody_columns:
+      let dc_identifier = 
+        DataColumnIdentifier(
+          block_root: blck.root,
+          index: i)
+      if dc_identifier notin quarantine.data_columns:
+        return false
+      else:
+        return true
+
+func dataColumnFetchRecord*(quarantine: DataColumnQuarantine,
+                            blck: electra.SignedBeaconBlock):
+                            DataColumnFetchRecord =
+  var indices: seq[ColumnIndex]
+  for i in quarantine.custody_columns:
+    let
+      idx = ColumnIndex(i)
+      dc_id = DataColumnIdentifier(
+        block_root: blck.root,
+        index: idx)
+    if not quarantine.data_columns.hasKey(
+        dc_id):
+      indices.add(idx)
+  DataColumnFetchRecord(block_root: blck.root, indices: indices)

beacon_chain/gossip_processing/gossip_validation.nim

@@ -15,10 +15,11 @@ import
   stew/byteutils,
   # Internals
   ../spec/[
-    beaconstate, state_transition_block, forks, helpers, network, signatures],
+    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
@@ -209,6 +210,22 @@ func check_blob_sidecar_inclusion_proof(
 
   ok()
 
+func check_data_column_sidecar_inclusion_proof(
+  data_column_sidecar: DataColumnSidecar): Result[void, ValidationError] =
+  let res = data_column_sidecar.verify_data_column_sidecar_inclusion_proof()
+  if res.isErr:
+    return errReject(res.error)
+
+  ok()
+
+proc check_data_column_sidecar_kzg_proofs(
+  data_column_sidecar: DataColumnSidecar): Result[void, ValidationError] =
+  let res = data_column_sidecar.verify_data_column_sidecar_kzg_proofs()
+  if res.isErr:
+    return errReject(res.error)
+
+  ok()
+
 # Gossip Validation
 # ----------------------------------------------------------------
 
@@ -475,6 +492,134 @@ proc validateBlobSidecar*(
 
   ok()
 
+# https://github.com/ethereum/consensus-specs/blob/v1.5.0-alpha.8/specs/_features/eip7594/p2p-interface.md#data_column_sidecar_subnet_id
+proc validateDataColumnSidecar*(
+    dag: ChainDAGRef, quarantine: ref Quarantine,
+    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
+
+  # [REJECT] The sidecar's index is consistent with `NUMBER_OF_COLUMNS`
+  # -- i.e. `data_column_sidecar.index < NUMBER_OF_COLUMNS`
+  if not (data_column_sidecar.index < NUMBER_OF_COLUMNS):
+    return dag.checkedReject("DataColumnSidecar: The sidecar's index should be consistent with NUMBER_OF_COLUMNS")
+
+  # [REJECT] The sidecar is for the correct subnet 
+  # -- i.e. `compute_subnet_for_data_column_sidecar(blob_sidecar.index) == subnet_id`.
+  if not (compute_subnet_for_data_column_sidecar(data_column_sidecar.index) == subnet_id):
+    return dag.checkedReject("DataColumnSidecar: The sidecar is not for the correct subnet")
+
+  # [IGNORE] The sidecar is not from a future slot 
+  # (with a `MAXIMUM_GOSSIP_CLOCK_DISPARITY` allowance) -- i.e. validate that 
+  # `block_header.slot <= current_slot`(a client MAY queue future sidecars for 
+  # processing at the appropriate slot).
+  if not (block_header.slot <=
+      (wallTime + MAXIMUM_GOSSIP_CLOCK_DISPARITY).slotOrZero):
+    return errIgnore("DataColumnSidecar: slot too high")
+
+  # [IGNORE] The sidecar is from a slot greater than the latest
+  # finalized slot -- i.e. validate that `block_header.slot >
+  # compute_start_slot_at_epoch(state.finalized_checkpoint.epoch)`
+  if not (block_header.slot > dag.finalizedHead.slot):
+    return errIgnore("DataColumnSidecar: slot already finalized")
+
+  # [IGNORE] The sidecar is the first sidecar for the tuple
+  # (block_header.slot, block_header.proposer_index, data_column_sidecar.index)
+  # with valid header signature, sidecar inclusion proof, and kzg proof.
+  let block_root = hash_tree_root(block_header)
+  if dag.getBlockRef(block_root).isSome():
+    return errIgnore("DataColumnSidecar: already have block")
+  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")
+
+  # [REJECT] The sidecar's `kzg_commitments` inclusion proof is valid as verified by
+  # `verify_data_column_sidecar_inclusion_proof(sidecar)`.
+  block:
+    let v = check_data_column_sidecar_inclusion_proof(data_column_sidecar)
+    if v.isErr:
+      return dag.checkedReject(v.error)
+
+  # [IGNORE] The sidecar's block's parent (defined by
+  # `block_header.parent_root`) has been seen (via both gossip and
+  # non-gossip sources) (a client MAY queue sidecars for processing
+  # once the parent block is retrieved).
+  #
+  # [REJECT] The sidecar's block's parent (defined by
+  # `block_header.parent_root`) passes validation.
+  let parent = dag.getBlockRef(block_header.parent_root).valueOr:
+    if block_header.parent_root in quarantine[].unviable:
+      quarantine[].addUnviable(block_root)
+      return dag.checkedReject("DataColumnSidecar: parent not validated")
+    else:
+      quarantine[].addMissing(block_header.parent_root)
+      return errIgnore("DataColumnSidecar: parent not found")
+
+  # [REJECT] The sidecar is from a higher slot than the sidecar's
+  # block's parent (defined by `block_header.parent_root`).
+  if not (block_header.slot > parent.bid.slot):
+    return dag.checkedReject("DataColumnSidecar: slot lower than parents'")
+
+  # [REJECT] The current finalized_checkpoint is an ancestor of the sidecar's
+  # block -- i.e. `get_checkpoint_block(store, block_header.parent_root,
+  # store.finalized_checkpoint.epoch) == store.finalized_checkpoint.root`.
+  let
+    finalized_checkpoint = getStateField(dag.headState, finalized_checkpoint)
+    ancestor = get_ancestor(parent, finalized_checkpoint.epoch.start_slot)
+
+  if ancestor.isNil:
+    # This shouldn't happen: we should always be able to trace the parent back
+    # to the finalized checkpoint (else it wouldn't be in the DAG)
+    return errIgnore("DataColumnSidecar: Can't find ancestor")
+
+  if not (
+      finalized_checkpoint.root == ancestor.root or
+      finalized_checkpoint.root.isZero):
+    quarantine[].addUnviable(block_root)
+    return dag.checkedReject(
+      "DataColumnSidecar: Finalized checkpoint not an ancestor")
+
+  # [REJECT] The sidecar is proposed by the expected `proposer_index`
+  # for the block's slot in the context of the current shuffling
+  # (defined by `block_header.parent_root`/`block_header.slot`).
+  # If the proposer_index cannot immediately be verified against the expected
+  # shuffling, the sidecar MAY be queued for later processing while proposers
+  # for the block's branch are calculated -- in such a case do not
+  # REJECT, instead IGNORE this message.
+  let proposer = getProposer(dag, parent, block_header.slot).valueOr:
+    warn "cannot compute proposer for data column"
+    return errIgnore("DataColumnSidecar: Cannot compute proposer") # internal issue
+
+  if uint64(proposer) != block_header.proposer_index:
+    return dag.checkedReject("DataColumnSidecar: Unexpected proposer")
+
+  # [REJECT] The proposer signature of `data_column_sidecar.signed_block_header`,
+  # is valid with respect to the `block_header.proposer_index` pubkey.
+  if not verify_block_signature(
+      dag.forkAtEpoch(block_header.slot.epoch),
+      getStateField(dag.headState, genesis_validators_root),
+      block_header.slot,
+      block_root,
+      dag.validatorKey(proposer).get(),
+      data_column_sidecar.signed_block_header.signature):
+    return dag.checkedReject("DataColumnSidecar: Invalid proposer signature")
+
+  # [REJECT] The sidecar's column data is valid as 
+  # verified by `verify_data_column_kzg_proofs(sidecar)`
+  block:
+    let r = check_data_column_sidecar_kzg_proofs(data_column_sidecar)
+    if r.isErr:
+      return dag.checkedReject(r.error)
+
+  # Send notification about new data column sidecar via callback
+  if not(isNil(dataColumnQuarantine.onDataColumnSidecarCallback)):
+    dataColumnQuarantine.onDataColumnSidecarCallback(data_column_sidecar)
+
+  ok()
+
 # https://github.com/ethereum/consensus-specs/blob/v1.3.0/specs/phase0/p2p-interface.md#beacon_block
 # https://github.com/ethereum/consensus-specs/blob/v1.3.0/specs/bellatrix/p2p-interface.md#beacon_block
 proc validateBeaconBlock*(