Implement PeerDAS Fulu fork activation (sigp#6795)

Squashed commit of the following:

commit b3da74b
Merge: e813532 a1b7d61
Author: Jimmy Chen <[email protected]>
Date:   Thu Jan 23 16:11:34 2025 +1100

    Merge remote-tracking branch 'origin/unstable' into jimmy/lh-2271-activate-peerdas-at-fulu-fork-and-remove-eip7594_fork_epoch

    # Conflicts:
    #	beacon_node/beacon_chain/src/fetch_blobs.rs
    #	beacon_node/store/src/lib.rs
    #	beacon_node/store/src/memory_store.rs

commit e813532
Author: Jimmy Chen <[email protected]>
Date:   Tue Jan 21 17:44:19 2025 +1100

    Skip blob pruning tests for Fulu.

commit 0e8f671
Merge: 614f984 33c1648
Author: Jimmy Chen <[email protected]>
Date:   Tue Jan 21 16:03:53 2025 +1100

    Merge branch 'unstable' into jimmy/lh-2271-activate-peerdas-at-fulu-fork-and-remove-eip7594_fork_epoch

commit 614f984
Author: Jimmy Chen <[email protected]>
Date:   Tue Jan 21 15:59:22 2025 +1100

    Fix range sync to select custody peers from its syncing chain instead of the global peer list.

commit eff9a5b
Author: Jimmy Chen <[email protected]>
Date:   Tue Jan 21 10:02:19 2025 +1100

    More test fixes for Fulu.

commit b63a6c4
Merge: b7da075 7a0388e
Author: Jimmy Chen <[email protected]>
Date:   Mon Jan 20 23:41:13 2025 +1100

    Merge remote-tracking branch 'origin/unstable' into jimmy/lh-2271-activate-peerdas-at-fulu-fork-and-remove-eip7594_fork_epoch

commit b7da075
Author: Jimmy Chen <[email protected]>
Date:   Mon Jan 20 16:03:36 2025 +1100

    More test fixes for Fulu.

commit 6d5b5ed
Merge: 8980832 06329ec
Author: Jimmy Chen <[email protected]>
Date:   Fri Jan 17 12:32:58 2025 +1100

    Merge remote-tracking branch 'origin/unstable' into jimmy/lh-2271-activate-peerdas-at-fulu-fork-and-remove-eip7594_fork_epoch

commit 8980832
Author: Jimmy Chen <[email protected]>
Date:   Fri Jan 17 11:41:46 2025 +1100

    Update Fulu spec tests. Revert back to testing Fulu as "feature", because all non-PeerDAS Fulu SSZ types are the same as Electra, and serde deserializes the vectors into Electra types.

commit 4d407fe
Merge: 8cdf82e b1a19a8
Author: Jimmy Chen <[email protected]>
Date:   Thu Jan 16 01:05:04 2025 +1100

    Merge remote-tracking branch 'origin/unstable' into jimmy/lh-2271-activate-peerdas-at-fulu-fork-and-remove-eip7594_fork_epoch

    # Conflicts:
    #	beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
    #	consensus/types/src/chain_spec.rs
    #	testing/ef_tests/src/cases.rs
    #	testing/ef_tests/src/cases/get_custody_groups.rs
    #	testing/ef_tests/src/cases/kzg_compute_cells_and_kzg_proofs.rs
    #	testing/ef_tests/src/cases/kzg_recover_cells_and_kzg_proofs.rs
    #	testing/ef_tests/src/cases/kzg_verify_cell_kzg_proof_batch.rs
    #	testing/ef_tests/src/handler.rs
    #	testing/ef_tests/tests/tests.rs

commit 8cdf82e
Author: Jimmy Chen <[email protected]>
Date:   Wed Jan 15 14:31:29 2025 +1100

    Use engine v4 methods for Fulu (v5 methods do not exist yet). Update kurtosis config for PeerDAS as electra genesis is not yet supported.

commit 4e25302
Author: Jimmy Chen <[email protected]>
Date:   Wed Jan 15 13:07:43 2025 +1100

    Address review comments and fix lint.

commit 0c9d64b
Merge: 64e44e1 587c3e2
Author: Jimmy Chen <[email protected]>
Date:   Wed Jan 15 12:48:27 2025 +1100

    Merge remote-tracking branch 'origin/unstable' into jimmy/lh-2271-activate-peerdas-at-fulu-fork-and-remove-eip7594_fork_epoch

    # Conflicts:
    #	beacon_node/lighthouse_network/src/rpc/protocol.rs
    #	testing/ef_tests/check_all_files_accessed.py
    #	testing/ef_tests/src/handler.rs

commit 64e44e1
Author: Jimmy Chen <[email protected]>
Date:   Tue Jan 14 14:45:09 2025 +1100

    Fix failing tests now `fulu` fork is included.

commit b029342
Author: Jimmy Chen <[email protected]>
Date:   Mon Jan 13 16:30:34 2025 +1100

    Fix compilation and update Kurtosis test config for PeerDAS.

commit cd77b2c
Author: Jimmy Chen <[email protected]>
Date:   Mon Jan 13 16:16:18 2025 +1100

    Update spec tests.

commit 2e11554
Author: Jimmy Chen <[email protected]>
Date:   Mon Jan 13 14:45:55 2025 +1100

    Implement PeerDAS Fulu fork activation.

Author: jimmygchen

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -1249,6 +1249,17 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         self.store.get_blobs(block_root).map_err(Error::from)
     }
 
+    /// Returns the data columns at the given root, if any.
+    ///
+    /// ## Errors
+    /// May return a database error.
+    pub fn get_data_columns(
+        &self,
+        block_root: &Hash256,
+    ) -> Result<Option<DataColumnSidecarList<T::EthSpec>>, Error> {
+        self.store.get_data_columns(block_root).map_err(Error::from)
+    }
+
     /// Returns the data columns at the given root, if any.
     ///
     /// ## Errors
@@ -5850,6 +5861,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
                 let kzg = self.kzg.as_ref();
 
+                // TODO(fulu): we no longer need blob proofs from PeerDAS and could avoid computing.
                 kzg_utils::validate_blobs::<T::EthSpec>(
                     kzg,
                     expected_kzg_commitments,

beacon_node/beacon_chain/src/data_column_verification.rs

@@ -699,7 +699,7 @@ mod test {
 
     #[tokio::test]
     async fn empty_data_column_sidecars_fails_validation() {
-        let spec = ForkName::latest().make_genesis_spec(E::default_spec());
+        let spec = ForkName::Fulu.make_genesis_spec(E::default_spec());
         let harness = BeaconChainHarness::builder(E::default())
             .spec(spec.into())
             .deterministic_keypairs(64)

beacon_node/beacon_chain/src/fulu_readiness.rs

@@ -1,7 +1,7 @@
 //! Provides tools for checking if a node is ready for the Fulu upgrade.
 
 use crate::{BeaconChain, BeaconChainTypes};
-use execution_layer::http::{ENGINE_GET_PAYLOAD_V5, ENGINE_NEW_PAYLOAD_V5};
+use execution_layer::http::{ENGINE_GET_PAYLOAD_V4, ENGINE_NEW_PAYLOAD_V4};
 use serde::{Deserialize, Serialize};
 use std::fmt;
 use std::time::Duration;
@@ -87,14 +87,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 Ok(capabilities) => {
                     let mut missing_methods = String::from("Required Methods Unsupported:");
                     let mut all_good = true;
-                    if !capabilities.get_payload_v5 {
+                    // TODO(fulu) switch to v5 when the EL is ready
+                    if !capabilities.get_payload_v4 {
                         missing_methods.push(' ');
-                        missing_methods.push_str(ENGINE_GET_PAYLOAD_V5);
+                        missing_methods.push_str(ENGINE_GET_PAYLOAD_V4);
                         all_good = false;
                     }
-                    if !capabilities.new_payload_v5 {
+                    if !capabilities.new_payload_v4 {
                         missing_methods.push(' ');
-                        missing_methods.push_str(ENGINE_NEW_PAYLOAD_V5);
+                        missing_methods.push_str(ENGINE_NEW_PAYLOAD_V4);
                         all_good = false;
                     }
 

beacon_node/beacon_chain/src/test_utils.rs

@@ -1,8 +1,9 @@
+use crate::blob_verification::GossipVerifiedBlob;
 use crate::block_verification_types::{AsBlock, RpcBlock};
+use crate::data_column_verification::CustodyDataColumn;
 use crate::kzg_utils::blobs_to_data_column_sidecars;
 use crate::observed_operations::ObservationOutcome;
 pub use crate::persisted_beacon_chain::PersistedBeaconChain;
-use crate::BeaconBlockResponseWrapper;
 pub use crate::{
     beacon_chain::{BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, FORK_CHOICE_DB_KEY, OP_POOL_DB_KEY},
     migrate::MigratorConfig,
@@ -16,6 +17,7 @@ use crate::{
     BeaconChain, BeaconChainTypes, BlockError, ChainConfig, ServerSentEventHandler,
     StateSkipConfig,
 };
+use crate::{get_block_root, BeaconBlockResponseWrapper};
 use bls::get_withdrawal_credentials;
 use eth2::types::SignedBlockContentsTuple;
 use execution_layer::test_utils::generate_genesis_header;
@@ -105,7 +107,7 @@ static KZG_NO_PRECOMP: LazyLock<Arc<Kzg>> = LazyLock::new(|| {
 });
 
 pub fn get_kzg(spec: &ChainSpec) -> Arc<Kzg> {
-    if spec.eip7594_fork_epoch.is_some() {
+    if spec.fulu_fork_epoch.is_some() {
         KZG_PEERDAS.clone()
     } else if spec.deneb_fork_epoch.is_some() {
         KZG.clone()
@@ -762,15 +764,13 @@ where
     pub fn get_head_block(&self) -> RpcBlock<E> {
         let block = self.chain.head_beacon_block();
         let block_root = block.canonical_root();
-        let blobs = self.chain.get_blobs(&block_root).unwrap().blobs();
-        RpcBlock::new(Some(block_root), block, blobs).unwrap()
+        self.build_rpc_block_from_store_blobs(Some(block_root), block)
     }
 
     pub fn get_full_block(&self, block_root: &Hash256) -> RpcBlock<E> {
         let block = self.chain.get_blinded_block(block_root).unwrap().unwrap();
         let full_block = self.chain.store.make_full_block(block_root, block).unwrap();
-        let blobs = self.chain.get_blobs(block_root).unwrap().blobs();
-        RpcBlock::new(Some(*block_root), Arc::new(full_block), blobs).unwrap()
+        self.build_rpc_block_from_store_blobs(Some(*block_root), Arc::new(full_block))
     }
 
     pub fn get_all_validators(&self) -> Vec<usize> {
@@ -2271,22 +2271,19 @@ where
         self.set_current_slot(slot);
         let (block, blob_items) = block_contents;
 
-        let sidecars = blob_items
-            .map(|(proofs, blobs)| BlobSidecar::build_sidecars(blobs, &block, proofs, &self.spec))
-            .transpose()
-            .unwrap();
+        let rpc_block = self.build_rpc_block_from_blobs(block_root, block, blob_items)?;
         let block_hash: SignedBeaconBlockHash = self
             .chain
             .process_block(
                 block_root,
-                RpcBlock::new(Some(block_root), block, sidecars).unwrap(),
+                rpc_block,
                 NotifyExecutionLayer::Yes,
                 BlockImportSource::RangeSync,
                 || Ok(()),
             )
             .await?
             .try_into()
-            .unwrap();
+            .expect("block blobs are available");
         self.chain.recompute_head_at_current_slot().await;
         Ok(block_hash)
     }
@@ -2297,16 +2294,13 @@ where
     ) -> Result<SignedBeaconBlockHash, BlockError> {
         let (block, blob_items) = block_contents;
 
-        let sidecars = blob_items
-            .map(|(proofs, blobs)| BlobSidecar::build_sidecars(blobs, &block, proofs, &self.spec))
-            .transpose()
-            .unwrap();
         let block_root = block.canonical_root();
+        let rpc_block = self.build_rpc_block_from_blobs(block_root, block, blob_items)?;
         let block_hash: SignedBeaconBlockHash = self
             .chain
             .process_block(
                 block_root,
-                RpcBlock::new(Some(block_root), block, sidecars).unwrap(),
+                rpc_block,
                 NotifyExecutionLayer::Yes,
                 BlockImportSource::RangeSync,
                 || Ok(()),
@@ -2318,6 +2312,82 @@ where
         Ok(block_hash)
     }
 
+    /// Builds an `Rpc` block from a `SignedBeaconBlock` and blobs or data columns retrieved from
+    /// the database.
+    pub fn build_rpc_block_from_store_blobs(
+        &self,
+        block_root: Option<Hash256>,
+        block: Arc<SignedBeaconBlock<E>>,
+    ) -> RpcBlock<E> {
+        let block_root = block_root.unwrap_or_else(|| get_block_root(&block));
+        let has_blobs = block
+            .message()
+            .body()
+            .blob_kzg_commitments()
+            .is_ok_and(|c| !c.is_empty());
+        if !has_blobs {
+            return RpcBlock::new_without_blobs(Some(block_root), block);
+        }
+
+        // Blobs are stored as data columns from Fulu (PeerDAS)
+        if self.spec.is_peer_das_enabled_for_epoch(block.epoch()) {
+            let columns = self.chain.get_data_columns(&block_root).unwrap().unwrap();
+            let custody_columns = columns
+                .into_iter()
+                .map(CustodyDataColumn::from_asserted_custody)
+                .collect::<Vec<_>>();
+            RpcBlock::new_with_custody_columns(Some(block_root), block, custody_columns, &self.spec)
+                .unwrap()
+        } else {
+            let blobs = self.chain.get_blobs(&block_root).unwrap().blobs();
+            RpcBlock::new(Some(block_root), block, blobs).unwrap()
+        }
+    }
+
+    /// Builds an `RpcBlock` from a `SignedBeaconBlock` and `BlobsList`.
+    fn build_rpc_block_from_blobs(
+        &self,
+        block_root: Hash256,
+        block: Arc<SignedBeaconBlock<E, FullPayload<E>>>,
+        blob_items: Option<(KzgProofs<E>, BlobsList<E>)>,
+    ) -> Result<RpcBlock<E>, BlockError> {
+        Ok(if self.spec.is_peer_das_enabled_for_epoch(block.epoch()) {
+            let sampling_column_count = self
+                .chain
+                .data_availability_checker
+                .get_sampling_column_count();
+
+            let columns = blob_items
+                .map(|(_proofs, blobs)| {
+                    blobs_to_data_column_sidecars(
+                        &blobs.iter().collect::<Vec<_>>(),
+                        &block,
+                        &self.chain.kzg,
+                        &self.spec,
+                    )
+                    .map(|column_sidecars| {
+                        column_sidecars
+                            .into_iter()
+                            .take(sampling_column_count)
+                            .map(CustodyDataColumn::from_asserted_custody)
+                            .collect::<Vec<_>>()
+                    })
+                })
+                .transpose()
+                .expect("should convert blobs to columns")
+                .unwrap_or_default();
+            RpcBlock::new_with_custody_columns(Some(block_root), block, columns, &self.spec)?
+        } else {
+            let blobs = blob_items
+                .map(|(proofs, blobs)| {
+                    BlobSidecar::build_sidecars(blobs, &block, proofs, &self.spec)
+                })
+                .transpose()
+                .unwrap();
+            RpcBlock::new(Some(block_root), block, blobs)?
+        })
+    }
+
     pub fn process_attestations(&self, attestations: HarnessAttestations<E>) {
         let num_validators = self.validator_keypairs.len();
         let mut unaggregated = Vec::with_capacity(num_validators);
@@ -2991,6 +3061,60 @@ where
 
         Ok(())
     }
+
+    /// Simulate some of the blobs / data columns being seen on gossip.
+    /// Converts the blobs to data columns if the slot is Fulu or later.
+    pub async fn process_gossip_blobs_or_columns<'a>(
+        &self,
+        block: &SignedBeaconBlock<E>,
+        blobs: impl Iterator<Item = &'a Blob<E>>,
+        proofs: impl Iterator<Item = &'a KzgProof>,
+        custody_columns_opt: Option<HashSet<ColumnIndex>>,
+    ) {
+        let is_peerdas_enabled = self.chain.spec.is_peer_das_enabled_for_epoch(block.epoch());
+        if is_peerdas_enabled {
+            let sidecars = blobs_to_data_column_sidecars(
+                &blobs.collect::<Vec<_>>(),
+                block,
+                &self.chain.kzg,
+                &self.spec,
+            )
+            .unwrap();
+
+            let custody_columns = custody_columns_opt.unwrap_or_else(|| {
+                let spec = &self.chain.spec;
+                let sampling_size = spec.sampling_size(spec.custody_requirement).unwrap();
+                (0..sampling_size).collect()
+            });
+
+            let verified_columns = sidecars
+                .into_iter()
+                .filter(|c| custody_columns.contains(&c.index))
+                .map(|sidecar| {
+                    let column_index = sidecar.index;
+                    self.chain
+                        .verify_data_column_sidecar_for_gossip(sidecar, column_index)
+                })
+                .collect::<Result<Vec<_>, _>>()
+                .unwrap();
+
+            self.chain
+                .process_gossip_data_columns(verified_columns, || Ok(()))
+                .await
+                .unwrap();
+        } else {
+            for (i, (kzg_proof, blob)) in proofs.into_iter().zip(blobs).enumerate() {
+                let sidecar =
+                    Arc::new(BlobSidecar::new(i, blob.clone(), block, *kzg_proof).unwrap());
+                let gossip_blob = GossipVerifiedBlob::new(sidecar, i as u64, &self.chain)
+                    .expect("should obtain gossip verified blob");
+                self.chain
+                    .process_gossip_blob(gossip_blob)
+                    .await
+                    .expect("should import valid gossip verified blob");
+            }
+        }
+    }
 }
 
 // Junk `Debug` impl to satistfy certain trait bounds during testing.

beacon_node/beacon_chain/tests/attestation_production.rs

@@ -1,7 +1,6 @@
 #![cfg(not(debug_assertions))]
 
 use beacon_chain::attestation_simulator::produce_unaggregated_attestation;
-use beacon_chain::block_verification_types::RpcBlock;
 use beacon_chain::test_utils::{AttestationStrategy, BeaconChainHarness, BlockStrategy};
 use beacon_chain::validator_monitor::UNAGGREGATED_ATTESTATION_LAG_SLOTS;
 use beacon_chain::{metrics, StateSkipConfig, WhenSlotSkipped};
@@ -155,7 +154,6 @@ async fn produces_attestations() {
             .store
             .make_full_block(&block_root, blinded_block)
             .unwrap();
-        let blobs = chain.get_blobs(&block_root).unwrap().blobs();
 
         let epoch_boundary_slot = state
             .current_epoch()
@@ -223,8 +221,7 @@ async fn produces_attestations() {
             assert_eq!(data.target.root, target_root, "bad target root");
 
             let rpc_block =
-                RpcBlock::<MainnetEthSpec>::new(None, Arc::new(block.clone()), blobs.clone())
-                    .unwrap();
+                harness.build_rpc_block_from_store_blobs(Some(block_root), Arc::new(block.clone()));
             let beacon_chain::data_availability_checker::MaybeAvailableBlock::Available(
                 available_block,
             ) = chain
@@ -296,14 +293,8 @@ async fn early_attester_cache_old_request() {
         .get_block(&head.beacon_block_root)
         .unwrap();
 
-    let head_blobs = harness
-        .chain
-        .get_blobs(&head.beacon_block_root)
-        .expect("should get blobs")
-        .blobs();
-
-    let rpc_block =
-        RpcBlock::<MainnetEthSpec>::new(None, head.beacon_block.clone(), head_blobs).unwrap();
+    let rpc_block = harness
+        .build_rpc_block_from_store_blobs(Some(head.beacon_block_root), head.beacon_block.clone());
     let beacon_chain::data_availability_checker::MaybeAvailableBlock::Available(available_block) =
         harness
             .chain