AgeManning
diff --git a/‎Cargo.lock
Lines changed: 2 additions & 0 deletions b/‎Cargo.lock
Lines changed: 2 additions & 0 deletions
diff --git a/‎beacon_node/beacon_chain/Cargo.toml
Lines changed: 5 additions & 0 deletions b/‎beacon_node/beacon_chain/Cargo.toml
Lines changed: 5 additions & 0 deletions
diff --git a/‎beacon_node/beacon_chain/benches/benches.rs
Lines changed: 66 additions & 0 deletions b/‎beacon_node/beacon_chain/benches/benches.rs
Lines changed: 66 additions & 0 deletions
diff --git a/‎beacon_node/beacon_chain/src/beacon_chain.rs
Lines changed: 89 additions & 28 deletions b/‎beacon_node/beacon_chain/src/beacon_chain.rs
Lines changed: 89 additions & 28 deletions
diff --git a/‎beacon_node/beacon_chain/src/blob_verification.rs
Lines changed: 2 additions & 2 deletions b/‎beacon_node/beacon_chain/src/blob_verification.rs
Lines changed: 2 additions & 2 deletions
@@ -5,6 +5,10 @@ authors = ["Paul Hauner <[email protected]>", "Age Manning <[email protected]
 edition = { workspace = true }
 autotests = false # using a single test binary compiles faster
 
+[[bench]]
+name = "benches"
+harness = false
+
 [features]
 default = ["participation_metrics"]
 write_ssz_files = []  # Writes debugging .ssz files to /tmp during block processing.
@@ -16,6 +20,7 @@ test_backfill = []
 [dev-dependencies]
 maplit = { workspace = true }
 serde_json = { workspace = true }
+criterion = { workspace = true }
 
 [dependencies]
 bitvec = { workspace = true }
 
@@ -0,0 +1,66 @@
+use std::sync::Arc;
+
+use beacon_chain::kzg_utils::{blobs_to_data_column_sidecars, reconstruct_data_columns};
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+
+use bls::Signature;
+use eth2_network_config::TRUSTED_SETUP_BYTES;
+use kzg::{Kzg, KzgCommitment, TrustedSetup};
+use types::{
+    beacon_block_body::KzgCommitments, BeaconBlock, BeaconBlockDeneb, Blob, BlobsList, ChainSpec,
+    EmptyBlock, EthSpec, MainnetEthSpec, SignedBeaconBlock,
+};
+
+fn create_test_block_and_blobs<E: EthSpec>(
+    num_of_blobs: usize,
+    spec: &ChainSpec,
+) -> (SignedBeaconBlock<E>, BlobsList<E>) {
+    let mut block = BeaconBlock::Deneb(BeaconBlockDeneb::empty(spec));
+    let mut body = block.body_mut();
+    let blob_kzg_commitments = body.blob_kzg_commitments_mut().unwrap();
+    *blob_kzg_commitments =
+        KzgCommitments::<E>::new(vec![KzgCommitment::empty_for_testing(); num_of_blobs]).unwrap();
+
+    let signed_block = SignedBeaconBlock::from_block(block, Signature::empty());
+
+    let blobs = (0..num_of_blobs)
+        .map(|_| Blob::<E>::default())
+        .collect::<Vec<_>>()
+        .into();
+
+    (signed_block, blobs)
+}
+
+fn all_benches(c: &mut Criterion) {
+    type E = MainnetEthSpec;
+    let spec = Arc::new(E::default_spec());
+
+    let trusted_setup: TrustedSetup = serde_json::from_reader(TRUSTED_SETUP_BYTES)
+        .map_err(|e| format!("Unable to read trusted setup file: {}", e))
+        .expect("should have trusted setup");
+    let kzg = Arc::new(Kzg::new_from_trusted_setup(trusted_setup).expect("should create kzg"));
+
+    for blob_count in [1, 2, 3, 6] {
+        let kzg = kzg.clone();
+        let (signed_block, blob_sidecars) = create_test_block_and_blobs::<E>(blob_count, &spec);
+
+        let column_sidecars =
+            blobs_to_data_column_sidecars(&blob_sidecars, &signed_block, &kzg.clone(), &spec)
+                .unwrap();
+
+        let spec = spec.clone();
+
+        c.bench_function(&format!("reconstruct_{}", blob_count), |b| {
+            b.iter(|| {
+                black_box(reconstruct_data_columns(
+                    &kzg,
+                    &column_sidecars.iter().as_slice()[0..column_sidecars.len() / 2],
+                    spec.as_ref(),
+                ))
+            })
+        });
+    }
+}
+
+criterion_group!(benches, all_benches);
+criterion_main!(benches);
@@ -22,6 +22,7 @@ pub use crate::canonical_head::CanonicalHead;
 use crate::chain_config::ChainConfig;
 use crate::data_availability_checker::{
     Availability, AvailabilityCheckError, AvailableBlock, DataAvailabilityChecker,
+    DataColumnsToPublish,
 };
 use crate::data_column_verification::{GossipDataColumnError, GossipVerifiedDataColumn};
 use crate::early_attester_cache::EarlyAttesterCache;
@@ -123,6 +124,7 @@ use task_executor::{ShutdownReason, TaskExecutor};
 use tokio_stream::Stream;
 use tree_hash::TreeHash;
 use types::blob_sidecar::FixedBlobSidecarList;
+use types::data_column_sidecar::{ColumnIndex, DataColumnIdentifier};
 use types::payload::BlockProductionVersion;
 use types::*;
 
@@ -206,11 +208,13 @@ impl TryInto<Hash256> for AvailabilityProcessingStatus {
 /// The result of a chain segment processing.
 pub enum ChainSegmentResult<E: EthSpec> {
     /// Processing this chain segment finished successfully.
-    Successful { imported_blocks: usize },
+    Successful {
+        imported_blocks: Vec<(Hash256, Slot)>,
+    },
     /// There was an error processing this chain segment. Before the error, some blocks could
     /// have been imported.
     Failed {
-        imported_blocks: usize,
+        imported_blocks: Vec<(Hash256, Slot)>,
         error: BlockError<E>,
     },
 }
@@ -2696,7 +2700,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         chain_segment: Vec<RpcBlock<T::EthSpec>>,
     ) -> Result<Vec<HashBlockTuple<T::EthSpec>>, ChainSegmentResult<T::EthSpec>> {
         // This function will never import any blocks.
-        let imported_blocks = 0;
+        let imported_blocks = vec![];
         let mut filtered_chain_segment = Vec::with_capacity(chain_segment.len());
 
         // Produce a list of the parent root and slot of the child of each block.
@@ -2802,7 +2806,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         chain_segment: Vec<RpcBlock<T::EthSpec>>,
         notify_execution_layer: NotifyExecutionLayer,
     ) -> ChainSegmentResult<T::EthSpec> {
-        let mut imported_blocks = 0;
+        let mut imported_blocks = vec![];
 
         // Filter uninteresting blocks from the chain segment in a blocking task.
         let chain = self.clone();
@@ -2862,6 +2866,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
             // Import the blocks into the chain.
             for signature_verified_block in signature_verified_blocks {
+                let block_slot = signature_verified_block.slot();
                 match self
                     .process_block(
                         signature_verified_block.block_root(),
@@ -2874,9 +2879,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 {
                     Ok(status) => {
                         match status {
-                            AvailabilityProcessingStatus::Imported(_) => {
+                            AvailabilityProcessingStatus::Imported(block_root) => {
                                 // The block was imported successfully.
-                                imported_blocks += 1;
+                                imported_blocks.push((block_root, block_slot));
                             }
                             AvailabilityProcessingStatus::MissingComponents(slot, block_root) => {
                                 warn!(self.log, "Blobs missing in response to range request";
@@ -2909,6 +2914,17 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         ChainSegmentResult::Successful { imported_blocks }
     }
 
+    /// Updates fork-choice node into a permanent `available` state so it can become a viable head.
+    /// Only completed sampling results are received. Blocks are unavailable by default and should
+    /// be pruned on finalization, on a timeout or by a max count.
+    pub async fn process_sampling_completed(self: &Arc<Self>, block_root: Hash256) {
+        // TODO(das): update fork-choice
+        // NOTE: It is possible that sampling complets before block is imported into fork choice,
+        // in that case we may need to update availability cache.
+        // TODO(das): These log levels are too high, reduce once DAS matures
+        info!(self.log, "Sampling completed"; "block_root" => %block_root);
+    }
+
     /// Returns `Ok(GossipVerifiedBlock)` if the supplied `block` should be forwarded onto the
     /// gossip network. The block is not imported into the chain, it is just partially verified.
     ///
@@ -2983,6 +2999,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Err(BlockError::BlockIsAlreadyKnown(blob.block_root()));
         }
 
+        // No need to process and import blobs beyond the PeerDAS epoch.
+        if self.spec.is_peer_das_enabled_for_epoch(blob.epoch()) {
+            return Err(BlockError::BlobNotRequired(blob.slot()));
+        }
+
         if let Some(event_handler) = self.event_handler.as_ref() {
             if event_handler.has_blob_sidecar_subscribers() {
                 event_handler.register(EventKind::BlobSidecar(SseBlobSidecar::from_blob_sidecar(
@@ -3000,7 +3021,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub async fn process_gossip_data_columns(
         self: &Arc<Self>,
         data_columns: Vec<GossipVerifiedDataColumn<T>>,
-    ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
+    ) -> Result<
+        (
+            AvailabilityProcessingStatus,
+            DataColumnsToPublish<T::EthSpec>,
+        ),
+        BlockError<T::EthSpec>,
+    > {
         let Ok((slot, block_root)) = data_columns
             .iter()
             .map(|c| (c.slot(), c.block_root()))
@@ -3067,7 +3094,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub async fn process_rpc_custody_columns(
         self: &Arc<Self>,
         custody_columns: DataColumnSidecarList<T::EthSpec>,
-    ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
+    ) -> Result<
+        (
+            AvailabilityProcessingStatus,
+            DataColumnsToPublish<T::EthSpec>,
+        ),
+        BlockError<T::EthSpec>,
+    > {
         let Ok((slot, block_root)) = custody_columns
             .iter()
             .map(|c| (c.slot(), c.block_root()))
@@ -3094,7 +3127,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let r = self
             .check_rpc_custody_columns_availability_and_import(slot, block_root, custody_columns)
             .await;
-        self.remove_notified(&block_root, r)
+        self.remove_notified_custody_columns(&block_root, r)
     }
 
     /// Remove any block components from the *processing cache* if we no longer require them. If the
@@ -3114,13 +3147,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
     /// Remove any block components from the *processing cache* if we no longer require them. If the
     /// block was imported full or erred, we no longer require them.
-    fn remove_notified_custody_columns(
+    fn remove_notified_custody_columns<P>(
         &self,
         block_root: &Hash256,
-        r: Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>>,
-    ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
-        let has_missing_components =
-            matches!(r, Ok(AvailabilityProcessingStatus::MissingComponents(_, _)));
+        r: Result<(AvailabilityProcessingStatus, P), BlockError<T::EthSpec>>,
+    ) -> Result<(AvailabilityProcessingStatus, P), BlockError<T::EthSpec>> {
+        let has_missing_components = matches!(
+            r,
+            Ok((AvailabilityProcessingStatus::MissingComponents(_, _), _))
+        );
         if !has_missing_components {
             self.reqresp_pre_import_cache.write().remove(block_root);
         }
@@ -3378,20 +3413,26 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         slot: Slot,
         block_root: Hash256,
         data_columns: Vec<GossipVerifiedDataColumn<T>>,
-    ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
+    ) -> Result<
+        (
+            AvailabilityProcessingStatus,
+            DataColumnsToPublish<T::EthSpec>,
+        ),
+        BlockError<T::EthSpec>,
+    > {
         if let Some(slasher) = self.slasher.as_ref() {
             for data_colum in &data_columns {
                 slasher.accept_block_header(data_colum.signed_block_header());
             }
         }
 
-        let availability = self.data_availability_checker.put_gossip_data_columns(
-            slot,
-            block_root,
-            data_columns,
-        )?;
+        let (availability, data_columns_to_publish) = self
+            .data_availability_checker
+            .put_gossip_data_columns(slot, block_root, data_columns)?;
 
-        self.process_availability(slot, availability).await
+        self.process_availability(slot, availability)
+            .await
+            .map(|result| (result, data_columns_to_publish))
     }
 
     /// Checks if the provided blobs can make any cached blocks available, and imports immediately
@@ -3440,7 +3481,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         slot: Slot,
         block_root: Hash256,
         custody_columns: DataColumnSidecarList<T::EthSpec>,
-    ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
+    ) -> Result<
+        (
+            AvailabilityProcessingStatus,
+            DataColumnsToPublish<T::EthSpec>,
+        ),
+        BlockError<T::EthSpec>,
+    > {
         // Need to scope this to ensure the lock is dropped before calling `process_availability`
         // Even an explicit drop is not enough to convince the borrow checker.
         {
@@ -3465,13 +3512,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         // This slot value is purely informative for the consumers of
         // `AvailabilityProcessingStatus::MissingComponents` to log an error with a slot.
-        let availability = self.data_availability_checker.put_rpc_custody_columns(
-            block_root,
-            slot.epoch(T::EthSpec::slots_per_epoch()),
-            custody_columns,
-        )?;
+        let (availability, data_columns_to_publish) =
+            self.data_availability_checker.put_rpc_custody_columns(
+                block_root,
+                slot.epoch(T::EthSpec::slots_per_epoch()),
+                custody_columns,
+            )?;
 
-        self.process_availability(slot, availability).await
+        self.process_availability(slot, availability)
+            .await
+            .map(|result| (result, data_columns_to_publish))
     }
 
     /// Imports a fully available block. Otherwise, returns `AvailabilityProcessingStatus::MissingComponents`
@@ -3522,6 +3572,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             );
         }
 
+        // TODO(das) record custody column available timestamp
+
         // import
         let chain = self.clone();
         let block_root = self
@@ -6895,6 +6947,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             && self.spec.is_peer_das_enabled_for_epoch(block_epoch)
     }
 
+    /// Returns true if we should issue a sampling request for this block
+    /// TODO(das): check if the block is still within the da_window
+    pub fn should_sample_slot(&self, slot: Slot) -> bool {
+        self.config.enable_sampling
+            && self
+                .spec
+                .is_peer_das_enabled_for_epoch(slot.epoch(T::EthSpec::slots_per_epoch()))
+    }
+
     pub fn logger(&self) -> &Logger {
         &self.log
     }
 
@@ -409,8 +409,8 @@ pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes>(
     // Verify that the blob_sidecar was received on the correct subnet.
     if blob_index != subnet {
         return Err(GossipBlobError::InvalidSubnet {
-            expected: blob_index,
-            received: subnet,
+            expected: subnet,
+            received: blob_index,
         });
     }