Send custody data column to DataAvailabilityChecker for determining…

… block importability (#5570)

* Only import custody data columns after publishing a block.

* Add `subscribe-all-data-column-subnets` and pass custody column count to `availability_cache`.

* Add custody requirement checks to `availability_cache`.

* Fix config not being passed to DAChecker and add more logging.

* Introduce `peer_das_epoch` and make blobs and columns mutually exclusive.

* Add DA filter for PeerDAS.

* Fix data availability check and use test_logger in tests.

* Fix subscribe to all data column subnets not working correctly.

* Fix tests.

* Only publish column sidecars if PeerDAS is activated. Add `PEER_DAS_EPOCH` chain spec serialization.

* Remove unused data column index in `OverflowKey`.

* Fix column sidecars incorrectly produced when there are no blobs.

* Re-instate index to `OverflowKey::DataColumn` and downgrade noisy debug log to `trace`.

Cargo.toml

@@ -119,6 +119,8 @@ libsecp256k1 = "0.7"
 log = "0.4"
 lru = "0.12"
 maplit = "1"
+mockall = "0.12"
+mockall_double = "0.3"
 num_cpus = "1"
 parking_lot = "0.12"
 paste = "1"

beacon_node/beacon_chain/src/block_verification.rs

@@ -48,7 +48,7 @@
 // returned alongside.
 #![allow(clippy::result_large_err)]
 
-use crate::blob_verification::{GossipBlobError, GossipVerifiedBlob};
+use crate::blob_verification::{GossipBlobError, GossipVerifiedBlob, GossipVerifiedBlobList};
 use crate::block_verification_types::{
     AsBlock, BlockContentsError, BlockImportData, GossipVerifiedBlockContents, RpcBlock,
 };
@@ -104,8 +104,8 @@ use tree_hash::TreeHash;
 use types::data_column_sidecar::DataColumnSidecarError;
 use types::{
     BeaconBlockRef, BeaconState, BeaconStateError, BlobSidecarList, ChainSpec, CloneConfig,
-    DataColumnSidecar, DataColumnSubnetId, Epoch, EthSpec, ExecutionBlockHash, Hash256,
-    InconsistentFork, PublicKey, PublicKeyBytes, RelativeEpoch, SignedBeaconBlock,
+    DataColumnSidecar, DataColumnSubnetId, Epoch, EthSpec, ExecutionBlockHash, FullPayload,
+    Hash256, InconsistentFork, PublicKey, PublicKeyBytes, RelativeEpoch, SignedBeaconBlock,
     SignedBeaconBlockHeader, Slot,
 };
 use types::{BlobSidecar, ExecPayload};
@@ -741,43 +741,16 @@ impl<T: BeaconChainTypes> IntoGossipVerifiedBlockContents<T> for PublishBlockReq
             })
             .transpose()?;
 
-        let gossip_verified_data_columns = gossip_verified_blobs
-            .as_ref()
-            .map(|blobs| {
-                // NOTE: we expect KZG to be initialized if the blobs are present
-                let kzg = chain
-                    .kzg
-                    .as_ref()
-                    .ok_or(BlockContentsError::DataColumnError(
-                        GossipDataColumnError::<T::EthSpec>::KzgNotInitialized,
-                    ))?;
-
-                let blob_sidecar_list: Vec<_> =
-                    blobs.iter().map(|blob| blob.clone_blob()).collect();
-                let blob_sidecar_list = BlobSidecarList::new(blob_sidecar_list)
-                    .map_err(DataColumnSidecarError::SszError)?;
-                let timer = metrics::start_timer(&metrics::DATA_COLUMN_SIDECAR_COMPUTATION);
-                let sidecars = DataColumnSidecar::build_sidecars(&blob_sidecar_list, &block, kzg)?;
-                drop(timer);
-                let mut gossip_verified_data_columns = vec![];
-                for sidecar in sidecars {
-                    let subnet = DataColumnSubnetId::try_from_column_index::<T::EthSpec>(
-                        sidecar.index as usize,
-                    )
-                    .map_err(|_| {
-                        BlockContentsError::<T::EthSpec>::DataColumnSidecarError(
-                            DataColumnSidecarError::DataColumnIndexOutOfBounds,
-                        )
-                    })?;
-                    let column = GossipVerifiedDataColumn::new(sidecar, subnet.into(), chain)?;
-                    gossip_verified_data_columns.push(column);
-                }
-                let gossip_verified_data_columns =
-                    GossipVerifiedDataColumnList::new(gossip_verified_data_columns)
-                        .map_err(DataColumnSidecarError::SszError)?;
-                Ok::<_, BlockContentsError<T::EthSpec>>(gossip_verified_data_columns)
-            })
-            .transpose()?;
+        let peer_das_enabled = chain
+            .spec
+            .peer_das_epoch
+            .map_or(false, |peer_das_epoch| block.epoch() >= peer_das_epoch);
+
+        let gossip_verified_data_columns = if peer_das_enabled {
+            build_gossip_verified_data_columns(chain, &block, gossip_verified_blobs.as_ref())?
+        } else {
+            None
+        };
 
         let gossip_verified_block = GossipVerifiedBlock::new(block, chain)?;
 
@@ -793,6 +766,49 @@ impl<T: BeaconChainTypes> IntoGossipVerifiedBlockContents<T> for PublishBlockReq
     }
 }
 
+fn build_gossip_verified_data_columns<T: BeaconChainTypes>(
+    chain: &BeaconChain<T>,
+    block: &SignedBeaconBlock<T::EthSpec, FullPayload<T::EthSpec>>,
+    gossip_verified_blobs: Option<&GossipVerifiedBlobList<T>>,
+) -> Result<Option<GossipVerifiedDataColumnList<T>>, BlockContentsError<T::EthSpec>> {
+    gossip_verified_blobs
+        // Only attempt to build data columns if blobs is non empty to avoid skewing the metrics.
+        .filter(|b| !b.is_empty())
+        .map(|blobs| {
+            // NOTE: we expect KZG to be initialized if the blobs are present
+            let kzg = chain
+                .kzg
+                .as_ref()
+                .ok_or(BlockContentsError::DataColumnError(
+                    GossipDataColumnError::<T::EthSpec>::KzgNotInitialized,
+                ))?;
+
+            let blob_sidecar_list: Vec<_> = blobs.iter().map(|blob| blob.clone_blob()).collect();
+            let blob_sidecar_list = BlobSidecarList::new(blob_sidecar_list)
+                .map_err(DataColumnSidecarError::SszError)?;
+            let timer = metrics::start_timer(&metrics::DATA_COLUMN_SIDECAR_COMPUTATION);
+            let sidecars = DataColumnSidecar::build_sidecars(&blob_sidecar_list, block, kzg)?;
+            drop(timer);
+            let mut gossip_verified_data_columns = vec![];
+            for sidecar in sidecars {
+                let subnet =
+                    DataColumnSubnetId::try_from_column_index::<T::EthSpec>(sidecar.index as usize)
+                        .map_err(|_| {
+                            BlockContentsError::<T::EthSpec>::DataColumnSidecarError(
+                                DataColumnSidecarError::DataColumnIndexOutOfBounds,
+                            )
+                        })?;
+                let column = GossipVerifiedDataColumn::new(sidecar, subnet.into(), chain)?;
+                gossip_verified_data_columns.push(column);
+            }
+            let gossip_verified_data_columns =
+                GossipVerifiedDataColumnList::new(gossip_verified_data_columns)
+                    .map_err(DataColumnSidecarError::SszError)?;
+            Ok::<_, BlockContentsError<T::EthSpec>>(gossip_verified_data_columns)
+        })
+        .transpose()
+}
+
 /// Implemented on types that can be converted into a `ExecutionPendingBlock`.
 ///
 /// Used to allow functions to accept blocks at various stages of verification.

beacon_node/beacon_chain/src/block_verification_types.rs

@@ -84,8 +84,8 @@ enum RpcBlockInner<E: EthSpec> {
     /// This variant is used with parent lookups and by-range responses. It should have all blobs
     /// ordered, all block roots matching, and the correct number of blobs for this block.
     BlockAndBlobs(Arc<SignedBeaconBlock<E>>, BlobSidecarList<E>),
-    /// This variant is used with parent lookups and by-range responses. It should have all data columns
-    /// ordered, all block roots matching, and the correct number of data columns for this block.
+    /// This variant is used with parent lookups and by-range responses. It should have all
+    /// requested data columns, all block roots matching for this block.
     BlockAndDataColumns(Arc<SignedBeaconBlock<E>>, DataColumnSidecarList<E>),
 }
 

beacon_node/beacon_chain/src/builder.rs

@@ -104,6 +104,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     kzg: Option<Arc<Kzg>>,
     task_executor: Option<TaskExecutor>,
     validator_monitor_config: Option<ValidatorMonitorConfig>,
+    import_all_data_columns: bool,
 }
 
 impl<TSlotClock, TEth1Backend, E, THotStore, TColdStore>
@@ -145,6 +146,7 @@ where
             kzg: None,
             task_executor: None,
             validator_monitor_config: None,
+            import_all_data_columns: false,
         }
     }
 
@@ -618,6 +620,12 @@ where
         self
     }
 
+    /// Sets whether to require and import all data columns when importing block.
+    pub fn import_all_data_columns(mut self, import_all_data_columns: bool) -> Self {
+        self.import_all_data_columns = import_all_data_columns;
+        self
+    }
+
     /// Sets the `BeaconChain` event handler backend.
     ///
     /// For example, provide `ServerSentEventHandler` as a `handler`.
@@ -968,8 +976,15 @@ where
             validator_monitor: RwLock::new(validator_monitor),
             genesis_backfill_slot,
             data_availability_checker: Arc::new(
-                DataAvailabilityChecker::new(slot_clock, self.kzg.clone(), store, &log, self.spec)
-                    .map_err(|e| format!("Error initializing DataAvailabiltyChecker: {:?}", e))?,
+                DataAvailabilityChecker::new(
+                    slot_clock,
+                    self.kzg.clone(),
+                    store,
+                    self.import_all_data_columns,
+                    &log,
+                    self.spec,
+                )
+                .map_err(|e| format!("Error initializing DataAvailabilityChecker: {:?}", e))?,
             ),
             kzg: self.kzg.clone(),
             block_production_state: Arc::new(Mutex::new(None)),