Remove CGC from data_availability checker (#7033)

- Part of #6767

Validator custody makes the CGC and set of sampling columns dynamic. Right now this information is stored twice:
- in the data availability checker
- in the network globals

If that state becomes dynamic we must make sure it is in sync updating it twice, or guarding it behind a mutex. However, I noted that we don't really have to keep the CGC inside the data availability checker. All consumers can actually read it from the network globals, and we can update `make_available` to read the expected count of data columns from the block.

Author: dapplion

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -3031,6 +3031,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub async fn verify_block_for_gossip(
         self: &Arc<Self>,
         block: Arc<SignedBeaconBlock<T::EthSpec>>,
+        custody_columns_count: usize,
     ) -> Result<GossipVerifiedBlock<T>, BlockError> {
         let chain = self.clone();
         self.task_executor
@@ -3040,7 +3041,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     let slot = block.slot();
                     let graffiti_string = block.message().body().graffiti().as_utf8_lossy();
 
-                    match GossipVerifiedBlock::new(block, &chain) {
+                    match GossipVerifiedBlock::new(block, &chain, custody_columns_count) {
                         Ok(verified) => {
                             let commitments_formatted = verified.block.commitments_formatted();
                             debug!(
@@ -7161,10 +7162,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         block_root: Hash256,
         block_data: AvailableBlockData<T::EthSpec>,
     ) -> Result<Option<StoreOp<T::EthSpec>>, String> {
-        // TODO(das) we currently store all subnet sampled columns. Tracking issue to exclude non
-        // custody columns: https://github.com/sigp/lighthouse/issues/6465
-        let _custody_columns_count = self.data_availability_checker.get_sampling_column_count();
-
         match block_data {
             AvailableBlockData::NoData => Ok(None),
             AvailableBlockData::Blobs(blobs) => {

beacon_node/beacon_chain/src/block_verification.rs

@@ -683,6 +683,7 @@ pub struct GossipVerifiedBlock<T: BeaconChainTypes> {
     pub block_root: Hash256,
     parent: Option<PreProcessingSnapshot<T::EthSpec>>,
     consensus_context: ConsensusContext<T::EthSpec>,
+    custody_columns_count: usize,
 }
 
 /// A wrapper around a `SignedBeaconBlock` that indicates that all signatures (except the deposit
@@ -718,6 +719,7 @@ pub trait IntoGossipVerifiedBlock<T: BeaconChainTypes>: Sized {
     fn into_gossip_verified_block(
         self,
         chain: &BeaconChain<T>,
+        custody_columns_count: usize,
     ) -> Result<GossipVerifiedBlock<T>, BlockError>;
     fn inner_block(&self) -> Arc<SignedBeaconBlock<T::EthSpec>>;
 }
@@ -726,6 +728,7 @@ impl<T: BeaconChainTypes> IntoGossipVerifiedBlock<T> for GossipVerifiedBlock<T>
     fn into_gossip_verified_block(
         self,
         _chain: &BeaconChain<T>,
+        _custody_columns_count: usize,
     ) -> Result<GossipVerifiedBlock<T>, BlockError> {
         Ok(self)
     }
@@ -738,8 +741,9 @@ impl<T: BeaconChainTypes> IntoGossipVerifiedBlock<T> for Arc<SignedBeaconBlock<T
     fn into_gossip_verified_block(
         self,
         chain: &BeaconChain<T>,
+        custody_columns_count: usize,
     ) -> Result<GossipVerifiedBlock<T>, BlockError> {
-        GossipVerifiedBlock::new(self, chain)
+        GossipVerifiedBlock::new(self, chain, custody_columns_count)
     }
 
     fn inner_block(&self) -> Arc<SignedBeaconBlock<T::EthSpec>> {
@@ -808,6 +812,7 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
     pub fn new(
         block: Arc<SignedBeaconBlock<T::EthSpec>>,
         chain: &BeaconChain<T>,
+        custody_columns_count: usize,
     ) -> Result<Self, BlockError> {
         // If the block is valid for gossip we don't supply it to the slasher here because
         // we assume it will be transformed into a fully verified block. We *do* need to supply
@@ -817,19 +822,22 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
         // The `SignedBeaconBlock` and `SignedBeaconBlockHeader` have the same canonical root,
         // but it's way quicker to calculate root of the header since the hash of the tree rooted
         // at `BeaconBlockBody` is already computed in the header.
-        Self::new_without_slasher_checks(block, &header, chain).map_err(|e| {
-            process_block_slash_info::<_, BlockError>(
-                chain,
-                BlockSlashInfo::from_early_error_block(header, e),
-            )
-        })
+        Self::new_without_slasher_checks(block, &header, chain, custody_columns_count).map_err(
+            |e| {
+                process_block_slash_info::<_, BlockError>(
+                    chain,
+                    BlockSlashInfo::from_early_error_block(header, e),
+                )
+            },
+        )
     }
 
     /// As for new, but doesn't pass the block to the slasher.
     fn new_without_slasher_checks(
         block: Arc<SignedBeaconBlock<T::EthSpec>>,
         block_header: &SignedBeaconBlockHeader,
         chain: &BeaconChain<T>,
+        custody_columns_count: usize,
     ) -> Result<Self, BlockError> {
         // Ensure the block is the correct structure for the fork at `block.slot()`.
         block
@@ -1036,6 +1044,7 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
             block_root,
             parent,
             consensus_context,
+            custody_columns_count,
         })
     }
 
@@ -1183,6 +1192,7 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
                 block: MaybeAvailableBlock::AvailabilityPending {
                     block_root: from.block_root,
                     block,
+                    custody_columns_count: from.custody_columns_count,
                 },
                 block_root: from.block_root,
                 parent: Some(parent),

beacon_node/beacon_chain/src/block_verification_types.rs

@@ -31,6 +31,7 @@ use types::{
 pub struct RpcBlock<E: EthSpec> {
     block_root: Hash256,
     block: RpcBlockInner<E>,
+    custody_columns_count: usize,
 }
 
 impl<E: EthSpec> Debug for RpcBlock<E> {
@@ -44,6 +45,10 @@ impl<E: EthSpec> RpcBlock<E> {
         self.block_root
     }
 
+    pub fn custody_columns_count(&self) -> usize {
+        self.custody_columns_count
+    }
+
     pub fn as_block(&self) -> &SignedBeaconBlock<E> {
         match &self.block {
             RpcBlockInner::Block(block) => block,
@@ -104,6 +109,8 @@ impl<E: EthSpec> RpcBlock<E> {
         Self {
             block_root,
             block: RpcBlockInner::Block(block),
+            // Block has zero columns
+            custody_columns_count: 0,
         }
     }
 
@@ -145,13 +152,16 @@ impl<E: EthSpec> RpcBlock<E> {
         Ok(Self {
             block_root,
             block: inner,
+            // Block is before PeerDAS
+            custody_columns_count: 0,
         })
     }
 
     pub fn new_with_custody_columns(
         block_root: Option<Hash256>,
         block: Arc<SignedBeaconBlock<E>>,
         custody_columns: Vec<CustodyDataColumn<E>>,
+        custody_columns_count: usize,
         spec: &ChainSpec,
     ) -> Result<Self, AvailabilityCheckError> {
         let block_root = block_root.unwrap_or_else(|| get_block_root(&block));
@@ -172,6 +182,7 @@ impl<E: EthSpec> RpcBlock<E> {
         Ok(Self {
             block_root,
             block: inner,
+            custody_columns_count,
         })
     }
 
@@ -239,10 +250,12 @@ impl<E: EthSpec> ExecutedBlock<E> {
             MaybeAvailableBlock::AvailabilityPending {
                 block_root: _,
                 block: pending_block,
+                custody_columns_count,
             } => Self::AvailabilityPending(AvailabilityPendingExecutedBlock::new(
                 pending_block,
                 import_data,
                 payload_verification_outcome,
+                custody_columns_count,
             )),
         }
     }
@@ -308,18 +321,21 @@ pub struct AvailabilityPendingExecutedBlock<E: EthSpec> {
     pub block: Arc<SignedBeaconBlock<E>>,
     pub import_data: BlockImportData<E>,
     pub payload_verification_outcome: PayloadVerificationOutcome,
+    pub custody_columns_count: usize,
 }
 
 impl<E: EthSpec> AvailabilityPendingExecutedBlock<E> {
     pub fn new(
         block: Arc<SignedBeaconBlock<E>>,
         import_data: BlockImportData<E>,
         payload_verification_outcome: PayloadVerificationOutcome,
+        custody_columns_count: usize,
     ) -> Self {
         Self {
             block,
             import_data,
             payload_verification_outcome,
+            custody_columns_count,
         }
     }
 
@@ -439,19 +455,13 @@ impl<E: EthSpec> AsBlock<E> for MaybeAvailableBlock<E> {
     fn as_block(&self) -> &SignedBeaconBlock<E> {
         match &self {
             MaybeAvailableBlock::Available(block) => block.as_block(),
-            MaybeAvailableBlock::AvailabilityPending {
-                block_root: _,
-                block,
-            } => block,
+            MaybeAvailableBlock::AvailabilityPending { block, .. } => block,
         }
     }
     fn block_cloned(&self) -> Arc<SignedBeaconBlock<E>> {
         match &self {
             MaybeAvailableBlock::Available(block) => block.block_cloned(),
-            MaybeAvailableBlock::AvailabilityPending {
-                block_root: _,
-                block,
-            } => block.clone(),
+            MaybeAvailableBlock::AvailabilityPending { block, .. } => block.clone(),
         }
     }
     fn canonical_root(&self) -> Hash256 {

beacon_node/beacon_chain/src/builder.rs

@@ -975,14 +975,8 @@ where
             validator_monitor: RwLock::new(validator_monitor),
             genesis_backfill_slot,
             data_availability_checker: Arc::new(
-                DataAvailabilityChecker::new(
-                    slot_clock,
-                    self.kzg.clone(),
-                    store,
-                    self.import_all_data_columns,
-                    self.spec,
-                )
-                .map_err(|e| format!("Error initializing DataAvailabilityChecker: {:?}", e))?,
+                DataAvailabilityChecker::new(slot_clock, self.kzg.clone(), store, self.spec)
+                    .map_err(|e| format!("Error initializing DataAvailabilityChecker: {:?}", e))?,
             ),
             kzg: self.kzg.clone(),
         };

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -111,21 +111,9 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         slot_clock: T::SlotClock,
         kzg: Arc<Kzg>,
         store: BeaconStore<T>,
-        import_all_data_columns: bool,
         spec: Arc<ChainSpec>,
     ) -> Result<Self, AvailabilityCheckError> {
-        let custody_group_count = spec.custody_group_count(import_all_data_columns);
-        // This should only panic if the chain spec contains invalid values.
-        let sampling_size = spec
-            .sampling_size(custody_group_count)
-            .expect("should compute node sampling size from valid chain spec");
-
-        let inner = DataAvailabilityCheckerInner::new(
-            OVERFLOW_LRU_CAPACITY,
-            store,
-            sampling_size as usize,
-            spec.clone(),
-        )?;
+        let inner = DataAvailabilityCheckerInner::new(OVERFLOW_LRU_CAPACITY, store, spec.clone())?;
         Ok(Self {
             availability_cache: Arc::new(inner),
             slot_clock,
@@ -134,14 +122,6 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         })
     }
 
-    pub fn get_sampling_column_count(&self) -> usize {
-        self.availability_cache.sampling_column_count()
-    }
-
-    pub(crate) fn is_supernode(&self) -> bool {
-        self.get_sampling_column_count() == self.spec.number_of_columns as usize
-    }
-
     /// Checks if the block root is currenlty in the availability cache awaiting import because
     /// of missing components.
     pub fn get_execution_valid_block(
@@ -326,6 +306,7 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         &self,
         block: RpcBlock<T::EthSpec>,
     ) -> Result<MaybeAvailableBlock<T::EthSpec>, AvailabilityCheckError> {
+        let custody_columns_count = block.custody_columns_count();
         let (block_root, block, blobs, data_columns) = block.deconstruct();
         if self.blobs_required_for_block(&block) {
             return if let Some(blob_list) = blobs {
@@ -339,7 +320,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                     spec: self.spec.clone(),
                 }))
             } else {
-                Ok(MaybeAvailableBlock::AvailabilityPending { block_root, block })
+                Ok(MaybeAvailableBlock::AvailabilityPending {
+                    block_root,
+                    block,
+                    custody_columns_count,
+                })
             };
         }
         if self.data_columns_required_for_block(&block) {
@@ -364,7 +349,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                     spec: self.spec.clone(),
                 }))
             } else {
-                Ok(MaybeAvailableBlock::AvailabilityPending { block_root, block })
+                Ok(MaybeAvailableBlock::AvailabilityPending {
+                    block_root,
+                    block,
+                    custody_columns_count,
+                })
             };
         }
 
@@ -421,6 +410,7 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         }
 
         for block in blocks {
+            let custody_columns_count = block.custody_columns_count();
             let (block_root, block, blobs, data_columns) = block.deconstruct();
 
             let maybe_available_block = if self.blobs_required_for_block(&block) {
@@ -433,7 +423,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                         spec: self.spec.clone(),
                     })
                 } else {
-                    MaybeAvailableBlock::AvailabilityPending { block_root, block }
+                    MaybeAvailableBlock::AvailabilityPending {
+                        block_root,
+                        block,
+                        custody_columns_count,
+                    }
                 }
             } else if self.data_columns_required_for_block(&block) {
                 if let Some(data_columns) = data_columns {
@@ -447,7 +441,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                         spec: self.spec.clone(),
                     })
                 } else {
-                    MaybeAvailableBlock::AvailabilityPending { block_root, block }
+                    MaybeAvailableBlock::AvailabilityPending {
+                        block_root,
+                        block,
+                        custody_columns_count,
+                    }
                 }
             } else {
                 MaybeAvailableBlock::Available(AvailableBlock {
@@ -804,6 +802,7 @@ pub enum MaybeAvailableBlock<E: EthSpec> {
     AvailabilityPending {
         block_root: Hash256,
         block: Arc<SignedBeaconBlock<E>>,
+        custody_columns_count: usize,
     },
 }
 

beacon_node/beacon_chain/src/data_availability_checker/error.rs

@@ -10,7 +10,7 @@ pub enum Error {
         blob_commitment: KzgCommitment,
         block_commitment: KzgCommitment,
     },
-    Unexpected(&'static str),
+    Unexpected(String),
     SszTypes(ssz_types::Error),
     MissingBlobs,
     MissingCustodyColumns,