Merge sigp/unstable into electra

Cargo.toml

@@ -9,6 +9,7 @@ members = [
     "beacon_node/client",
     "beacon_node/eth1",
     "beacon_node/lighthouse_network",
+    "beacon_node/lighthouse_network/gossipsub",
     "beacon_node/execution_layer",
     "beacon_node/http_api",
     "beacon_node/http_metrics",
@@ -200,6 +201,7 @@ execution_layer = { path = "beacon_node/execution_layer" }
 filesystem = { path = "common/filesystem" }
 fork_choice = { path = "consensus/fork_choice" }
 genesis = { path = "beacon_node/genesis" }
+gossipsub = { path = "beacon_node/lighthouse_network/gossipsub/" }
 http_api = { path = "beacon_node/http_api" }
 int_to_bytes = { path = "consensus/int_to_bytes" }
 kzg = { path = "crypto/kzg" }

beacon_node/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "beacon_node"
-version = "5.1.1"
+version = "5.1.3"
 authors = [
     "Paul Hauner <paul@paulhauner.com>",
     "Age Manning <Age@AgeManning.com",

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -1347,11 +1347,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         (parent_root, slot, sync_aggregate): LightClientProducerEvent<T::EthSpec>,
     ) -> Result<(), Error> {
         self.light_client_server_cache.recompute_and_cache_updates(
-            &self.log,
             self.store.clone(),
             &parent_root,
             slot,
             &sync_aggregate,
+            &self.log,
+            &self.spec,
         )
     }
 
@@ -2566,7 +2567,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         &self,
         epoch: Epoch,
         validator_indices: &[u64],
-    ) -> Result<Vec<Option<SyncDuty>>, Error> {
+    ) -> Result<Vec<Result<Option<SyncDuty>, BeaconStateError>>, Error> {
         self.with_head(move |head| {
             head.beacon_state
                 .get_sync_committee_duties(epoch, validator_indices, &self.spec)
@@ -2651,7 +2652,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 // If the block is relevant, add it to the filtered chain segment.
                 Ok(_) => filtered_chain_segment.push((block_root, block)),
                 // If the block is already known, simply ignore this block.
-                Err(BlockError::BlockIsAlreadyKnown) => continue,
+                Err(BlockError::BlockIsAlreadyKnown(_)) => continue,
                 // If the block is the genesis block, simply ignore this block.
                 Err(BlockError::GenesisBlock) => continue,
                 // If the block is is for a finalized slot, simply ignore this block.
@@ -2795,6 +2796,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                             }
                         }
                     }
+                    Err(BlockError::BlockIsAlreadyKnown(block_root)) => {
+                        debug!(self.log,
+                            "Ignoring already known blocks while processing chain segment";
+                            "block_root" => ?block_root);
+                        continue;
+                    }
                     Err(error) => {
                         return ChainSegmentResult::Failed {
                             imported_blocks,
@@ -2879,7 +2886,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .fork_choice_read_lock()
             .contains_block(&block_root)
         {
-            return Err(BlockError::BlockIsAlreadyKnown);
+            return Err(BlockError::BlockIsAlreadyKnown(blob.block_root()));
         }
 
         if let Some(event_handler) = self.event_handler.as_ref() {
@@ -2891,7 +2898,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
 
         self.data_availability_checker
-            .notify_gossip_blob(blob.slot(), block_root, &blob);
+            .notify_gossip_blob(block_root, &blob);
         let r = self.check_gossip_blob_availability_and_import(blob).await;
         self.remove_notified(&block_root, r)
     }
@@ -2911,7 +2918,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .fork_choice_read_lock()
             .contains_block(&block_root)
         {
-            return Err(BlockError::BlockIsAlreadyKnown);
+            return Err(BlockError::BlockIsAlreadyKnown(block_root));
         }
 
         if let Some(event_handler) = self.event_handler.as_ref() {
@@ -2925,7 +2932,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
 
         self.data_availability_checker
-            .notify_rpc_blobs(slot, block_root, &blobs);
+            .notify_rpc_blobs(block_root, &blobs);
         let r = self
             .check_rpc_blob_availability_and_import(slot, block_root, blobs)
             .await;
@@ -3031,7 +3038,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         match import_block.await {
             // The block was successfully verified and imported. Yay.
             Ok(status @ AvailabilityProcessingStatus::Imported(block_root)) => {
-                trace!(
+                debug!(
                     self.log,
                     "Beacon block imported";
                     "block_root" => ?block_root,
@@ -3044,7 +3051,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 Ok(status)
             }
             Ok(status @ AvailabilityProcessingStatus::MissingComponents(slot, block_root)) => {
-                trace!(
+                debug!(
                     self.log,
                     "Beacon block awaiting blobs";
                     "block_root" => ?block_root,
@@ -6675,13 +6682,17 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         &self,
         block_root: &Hash256,
     ) -> Result<Option<(LightClientBootstrap<T::EthSpec>, ForkName)>, Error> {
-        let Some((state_root, slot)) = self
-            .get_blinded_block(block_root)?
-            .map(|block| (block.state_root(), block.slot()))
-        else {
+        let handle = self
+            .task_executor
+            .handle()
+            .ok_or(BeaconChainError::RuntimeShutdown)?;
+
+        let Some(block) = handle.block_on(async { self.get_block(block_root).await })? else {
             return Ok(None);
         };
 
+        let (state_root, slot) = (block.state_root(), block.slot());
+
         let Some(mut state) = self.get_state(&state_root, Some(slot))? else {
             return Ok(None);
         };
@@ -6691,14 +6702,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .map_err(Error::InconsistentFork)?;
 
         match fork_name {
-            ForkName::Altair | ForkName::Merge => {
-                LightClientBootstrap::from_beacon_state(&mut state)
+            ForkName::Altair
+            | ForkName::Merge
+            | ForkName::Capella
+            | ForkName::Deneb
+            | ForkName::Electra => {
+                LightClientBootstrap::from_beacon_state(&mut state, &block, &self.spec)
                     .map(|bootstrap| Some((bootstrap, fork_name)))
                     .map_err(Error::LightClientError)
             }
-            ForkName::Base | ForkName::Capella | ForkName::Deneb | ForkName::Electra => {
-                Err(Error::UnsupportedFork)
-            }
+            ForkName::Base => Err(Error::UnsupportedFork),
         }
     }
 }

beacon_node/beacon_chain/src/block_verification.rs

@@ -190,7 +190,7 @@ pub enum BlockError<T: EthSpec> {
     /// ## Peer scoring
     ///
     /// The block is valid and we have already imported a block with this hash.
-    BlockIsAlreadyKnown,
+    BlockIsAlreadyKnown(Hash256),
     /// The block slot exceeds the MAXIMUM_BLOCK_SLOT_NUMBER.
     ///
     /// ## Peer scoring
@@ -832,7 +832,7 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
         // already know this block.
         let fork_choice_read_lock = chain.canonical_head.fork_choice_read_lock();
         if fork_choice_read_lock.contains_block(&block_root) {
-            return Err(BlockError::BlockIsAlreadyKnown);
+            return Err(BlockError::BlockIsAlreadyKnown(block_root));
         }
 
         // Do not process a block that doesn't descend from the finalized root.
@@ -966,7 +966,7 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
             SeenBlock::Slashable => {
                 return Err(BlockError::Slashable);
             }
-            SeenBlock::Duplicate => return Err(BlockError::BlockIsAlreadyKnown),
+            SeenBlock::Duplicate => return Err(BlockError::BlockIsAlreadyKnown(block_root)),
             SeenBlock::UniqueNonSlashable => {}
         };
 
@@ -1784,7 +1784,7 @@ pub fn check_block_relevancy<T: BeaconChainTypes>(
         .fork_choice_read_lock()
         .contains_block(&block_root)
     {
-        return Err(BlockError::BlockIsAlreadyKnown);
+        return Err(BlockError::BlockIsAlreadyKnown(block_root));
     }
 
     Ok(block_root)

beacon_node/beacon_chain/src/block_verification_types.rs

@@ -7,6 +7,7 @@ use crate::{get_block_root, GossipVerifiedBlock, PayloadVerificationOutcome};
 use derivative::Derivative;
 use ssz_types::VariableList;
 use state_processing::ConsensusContext;
+use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 use types::blob_sidecar::{BlobIdentifier, BlobSidecarError, FixedBlobSidecarList};
 use types::{
@@ -27,13 +28,19 @@ use types::{
 /// Note: We make a distinction over blocks received over gossip because
 /// in a post-deneb world, the blobs corresponding to a given block that are received
 /// over rpc do not contain the proposer signature for dos resistance.
-#[derive(Debug, Clone, Derivative)]
+#[derive(Clone, Derivative)]
 #[derivative(Hash(bound = "E: EthSpec"))]
 pub struct RpcBlock<E: EthSpec> {
     block_root: Hash256,
     block: RpcBlockInner<E>,
 }
 
+impl<E: EthSpec> Debug for RpcBlock<E> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "RpcBlock({:?})", self.block_root)
+    }
+}
+
 impl<E: EthSpec> RpcBlock<E> {
     pub fn block_root(&self) -> Hash256 {
         self.block_root

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -22,7 +22,7 @@ use std::sync::Arc;
 use task_executor::TaskExecutor;
 use types::beacon_block_body::KzgCommitmentOpts;
 use types::blob_sidecar::{BlobIdentifier, BlobSidecar, FixedBlobSidecarList};
-use types::{BlobSidecarList, ChainSpec, Epoch, EthSpec, Hash256, SignedBeaconBlock, Slot};
+use types::{BlobSidecarList, ChainSpec, Epoch, EthSpec, Hash256, SignedBeaconBlock};
 
 mod availability_view;
 mod child_components;
@@ -110,8 +110,6 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         self.processing_cache.read().get(&block_root).cloned()
     }
 
-    /// A `None` indicates blobs are not required.
-    ///
     /// If there's no block, all possible ids will be returned that don't exist in the given blobs.
     /// If there no blobs, all possible ids will be returned.
     pub fn get_missing_blob_ids<V: AvailabilityView<T::EthSpec>>(
@@ -356,41 +354,30 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
     /// them here is useful to avoid duplicate downloads of blocks, as well as understanding
     /// our blob download requirements. We will also serve this over RPC.
     pub fn notify_block(&self, block_root: Hash256, block: Arc<SignedBeaconBlock<T::EthSpec>>) {
-        let slot = block.slot();
         self.processing_cache
             .write()
             .entry(block_root)
-            .or_insert_with(|| ProcessingComponents::new(slot))
+            .or_default()
             .merge_block(block);
     }
 
     /// Add a single blob commitment to the processing cache. This commitment is unverified but caching
     /// them here is useful to avoid duplicate downloads of blobs, as well as understanding
     /// our block and blob download requirements.
-    pub fn notify_gossip_blob(
-        &self,
-        slot: Slot,
-        block_root: Hash256,
-        blob: &GossipVerifiedBlob<T>,
-    ) {
+    pub fn notify_gossip_blob(&self, block_root: Hash256, blob: &GossipVerifiedBlob<T>) {
         let index = blob.index();
         let commitment = blob.kzg_commitment();
         self.processing_cache
             .write()
             .entry(block_root)
-            .or_insert_with(|| ProcessingComponents::new(slot))
+            .or_default()
             .merge_single_blob(index as usize, commitment);
     }
 
     /// Adds blob commitments to the processing cache. These commitments are unverified but caching
     /// them here is useful to avoid duplicate downloads of blobs, as well as understanding
     /// our block and blob download requirements.
-    pub fn notify_rpc_blobs(
-        &self,
-        slot: Slot,
-        block_root: Hash256,
-        blobs: &FixedBlobSidecarList<T::EthSpec>,
-    ) {
+    pub fn notify_rpc_blobs(&self, block_root: Hash256, blobs: &FixedBlobSidecarList<T::EthSpec>) {
         let mut commitments = KzgCommitmentOpts::<T::EthSpec>::default();
         for blob in blobs.iter().flatten() {
             if let Some(commitment) = commitments.get_mut(blob.index as usize) {
@@ -400,7 +387,7 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         self.processing_cache
             .write()
             .entry(block_root)
-            .or_insert_with(|| ProcessingComponents::new(slot))
+            .or_default()
             .merge_blobs(commitments);
     }
 
@@ -409,14 +396,6 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         self.processing_cache.write().remove(block_root)
     }
 
-    /// Gather all block roots for which we are not currently processing all components for the
-    /// given slot.
-    pub fn incomplete_processing_components(&self, slot: Slot) -> Vec<Hash256> {
-        self.processing_cache
-            .read()
-            .incomplete_processing_components(slot)
-    }
-
     /// The epoch at which we require a data availability check in block processing.
     /// `None` if the `Deneb` fork is disabled.
     pub fn data_availability_boundary(&self) -> Option<Epoch> {

beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs

@@ -108,11 +108,10 @@ pub trait AvailabilityView<E: EthSpec> {
     /// 1. The blob entry at the index is empty and no block exists, or
     /// 2. The block exists and its commitment matches the blob's commitment.
     fn merge_single_blob(&mut self, index: usize, blob: Self::BlobType) {
-        let commitment = *blob.get_commitment();
         if let Some(cached_block) = self.get_cached_block() {
             let block_commitment_opt = cached_block.get_commitments().get(index).copied();
             if let Some(block_commitment) = block_commitment_opt {
-                if block_commitment == commitment {
+                if block_commitment == *blob.get_commitment() {
                     self.insert_blob_at_index(index, blob)
                 }
             }

beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs

@@ -3,7 +3,7 @@ use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::sync::Arc;
 use types::beacon_block_body::KzgCommitmentOpts;
-use types::{EthSpec, Hash256, SignedBeaconBlock, Slot};
+use types::{EthSpec, Hash256, SignedBeaconBlock};
 
 /// This cache is used only for gossip blocks/blobs and single block/blob lookups, to give req/resp
 /// a view of what we have and what we require. This cache serves a slightly different purpose than
@@ -29,23 +29,13 @@ impl<E: EthSpec> ProcessingCache<E> {
             .get(block_root)
             .map_or(false, |b| b.block_exists())
     }
-    pub fn incomplete_processing_components(&self, slot: Slot) -> Vec<Hash256> {
-        let mut roots_missing_components = vec![];
-        for (&block_root, info) in self.processing_cache.iter() {
-            if info.slot == slot && !info.is_available() {
-                roots_missing_components.push(block_root);
-            }
-        }
-        roots_missing_components
-    }
     pub fn len(&self) -> usize {
         self.processing_cache.len()
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Default, Debug, Clone)]
 pub struct ProcessingComponents<E: EthSpec> {
-    slot: Slot,
     /// Blobs required for a block can only be known if we have seen the block. So `Some` here
     /// means we've seen it, a `None` means we haven't. The `kzg_commitments` value helps us figure
     /// out whether incoming blobs actually match the block.
@@ -56,12 +46,8 @@ pub struct ProcessingComponents<E: EthSpec> {
 }
 
 impl<E: EthSpec> ProcessingComponents<E> {
-    pub fn new(slot: Slot) -> Self {
-        Self {
-            slot,
-            block: None,
-            blob_commitments: KzgCommitmentOpts::<E>::default(),
-        }
+    pub fn new() -> Self {
+        Self::default()
     }
 }
 
@@ -70,7 +56,6 @@ impl<E: EthSpec> ProcessingComponents<E> {
 impl<E: EthSpec> ProcessingComponents<E> {
     pub fn empty(_block_root: Hash256) -> Self {
         Self {
-            slot: Slot::new(0),
             block: None,
             blob_commitments: KzgCommitmentOpts::<E>::default(),
         }

beacon_node/beacon_chain/src/errors.rs

@@ -250,6 +250,7 @@ easy_from_to!(StateAdvanceError, BeaconChainError);
 easy_from_to!(BlockReplayError, BeaconChainError);
 easy_from_to!(InconsistentFork, BeaconChainError);
 easy_from_to!(AvailabilityCheckError, BeaconChainError);
+easy_from_to!(LightClientError, BeaconChainError);
 
 #[derive(Debug)]
 pub enum BlockProductionError {