Add Electra forks to basic sim tests

.github/workflows/test-suite.yml

@@ -295,8 +295,8 @@ jobs:
       with:
           channel: stable
           cache-target: release
-    - name: Run a basic beacon chain sim that starts from Bellatrix
-      run: cargo run --release --bin simulator basic-sim
+    - name: Run a basic beacon chain sim that starts from Deneb
+      run: cargo run --release --bin simulator basic-sim --speed-up-factor 2
   fallback-simulator-ubuntu:
     name: fallback-simulator-ubuntu
     needs: [check-labels]

Cargo.toml

@@ -161,6 +161,7 @@ maplit = "1"
 milhouse = "0.5"
 mockito = "1.5.0"
 num_cpus = "1"
+once_cell = "1.17.1"
 parking_lot = "0.12"
 paste = "1"
 prometheus = { version = "0.13", default-features = false }

beacon_node/beacon_chain/Cargo.toml

@@ -47,6 +47,7 @@ logging = { workspace = true }
 lru = { workspace = true }
 merkle_proof = { workspace = true }
 metrics = { workspace = true }
+once_cell = { workspace = true }
 oneshot_broadcast = { path = "../../common/oneshot_broadcast/" }
 operation_pool = { workspace = true }
 parking_lot = { workspace = true }

beacon_node/beacon_chain/src/beacon_proposer_cache.rs

@@ -11,10 +11,12 @@
 use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
 use fork_choice::ExecutionStatus;
 use lru::LruCache;
+use once_cell::sync::OnceCell;
 use smallvec::SmallVec;
 use state_processing::state_advance::partial_state_advance;
 use std::cmp::Ordering;
 use std::num::NonZeroUsize;
+use std::sync::Arc;
 use types::non_zero_usize::new_non_zero_usize;
 use types::{
     BeaconState, BeaconStateError, ChainSpec, Epoch, EthSpec, Fork, Hash256, Slot, Unsigned,
@@ -39,21 +41,21 @@ pub struct Proposer {
 /// their signatures.
 pub struct EpochBlockProposers {
     /// The epoch to which the proposers pertain.
-    epoch: Epoch,
+    pub(crate) epoch: Epoch,
     /// The fork that should be used to verify proposer signatures.
-    fork: Fork,
+    pub(crate) fork: Fork,
     /// A list of length `T::EthSpec::slots_per_epoch()`, representing the proposers for each slot
     /// in that epoch.
     ///
     /// E.g., if `self.epoch == 1`, then `self.proposers[0]` contains the proposer for slot `32`.
-    proposers: SmallVec<[usize; TYPICAL_SLOTS_PER_EPOCH]>,
+    pub(crate) proposers: SmallVec<[usize; TYPICAL_SLOTS_PER_EPOCH]>,
 }
 
 /// A cache to store the proposers for some epoch.
 ///
 /// See the module-level documentation for more information.
 pub struct BeaconProposerCache {
-    cache: LruCache<(Epoch, Hash256), EpochBlockProposers>,
+    cache: LruCache<(Epoch, Hash256), Arc<OnceCell<EpochBlockProposers>>>,
 }
 
 impl Default for BeaconProposerCache {
@@ -74,7 +76,8 @@ impl BeaconProposerCache {
     ) -> Option<Proposer> {
         let epoch = slot.epoch(E::slots_per_epoch());
         let key = (epoch, shuffling_decision_block);
-        if let Some(cache) = self.cache.get(&key) {
+        let cache_opt = self.cache.get(&key).and_then(|cell| cell.get());
+        if let Some(cache) = cache_opt {
             // This `if` statement is likely unnecessary, but it feels like good practice.
             if epoch == cache.epoch {
                 cache
@@ -103,7 +106,26 @@ impl BeaconProposerCache {
         epoch: Epoch,
     ) -> Option<&SmallVec<[usize; TYPICAL_SLOTS_PER_EPOCH]>> {
         let key = (epoch, shuffling_decision_block);
-        self.cache.get(&key).map(|cache| &cache.proposers)
+        self.cache
+            .get(&key)
+            .and_then(|cache_once_cell| cache_once_cell.get().map(|proposers| &proposers.proposers))
+    }
+
+    /// Returns the `OnceCell` for the given `(epoch, shuffling_decision_block)` key,
+    /// inserting an empty one if it doesn't exist.
+    ///
+    /// The returned `OnceCell` allows the caller to initialise the value externally
+    /// using `get_or_try_init`, enabling deferred computation without holding a mutable
+    /// reference to the cache.
+    pub fn get_or_insert_key(
+        &mut self,
+        epoch: Epoch,
+        shuffling_decision_block: Hash256,
+    ) -> Arc<OnceCell<EpochBlockProposers>> {
+        let key = (epoch, shuffling_decision_block);
+        self.cache
+            .get_or_insert(key, || Arc::new(OnceCell::new()))
+            .clone()
     }
 
     /// Insert the proposers into the cache.
@@ -120,14 +142,13 @@ impl BeaconProposerCache {
     ) -> Result<(), BeaconStateError> {
         let key = (epoch, shuffling_decision_block);
         if !self.cache.contains(&key) {
-            self.cache.put(
-                key,
-                EpochBlockProposers {
-                    epoch,
-                    fork,
-                    proposers: proposers.into(),
-                },
-            );
+            let epoch_proposers = EpochBlockProposers {
+                epoch,
+                fork,
+                proposers: proposers.into(),
+            };
+            self.cache
+                .put(key, Arc::new(OnceCell::with_value(epoch_proposers)));
         }
 
         Ok(())

beacon_node/beacon_chain/src/data_column_verification.rs

@@ -1,3 +1,4 @@
+use crate::beacon_proposer_cache::EpochBlockProposers;
 use crate::block_verification::{
     cheap_state_advance_to_obtain_committees, get_validator_pubkey_cache, process_block_slash_info,
     BlockSlashInfo,
@@ -602,14 +603,19 @@ fn verify_proposer_and_signature<T: BeaconChainTypes>(
             parent_block.root
         };
 
-    let proposer_opt = chain
+    // We lock the cache briefly to get or insert a OnceCell, then drop the lock
+    // before doing proposer shuffling calculation via `OnceCell::get_or_try_init`. This avoids
+    // holding the lock during the computation, while still ensuring the result is cached and
+    // initialised only once.
+    //
+    // This approach exposes the cache internals (`OnceCell` & `EpochBlockProposers`)
+    //  as a trade-off for avoiding lock contention.
+    let epoch_proposers_cell = chain
         .beacon_proposer_cache
         .lock()
-        .get_slot::<T::EthSpec>(proposer_shuffling_root, column_slot);
+        .get_or_insert_key(column_epoch, proposer_shuffling_root);
 
-    let (proposer_index, fork) = if let Some(proposer) = proposer_opt {
-        (proposer.index, proposer.fork)
-    } else {
+    let epoch_proposers = epoch_proposers_cell.get_or_try_init(move || {
         debug!(
             %block_root,
             index = %column_index,
@@ -633,19 +639,20 @@ fn verify_proposer_and_signature<T: BeaconChainTypes>(
         )?;
 
         let proposers = state.get_beacon_proposer_indices(&chain.spec)?;
-        let proposer_index = *proposers
-            .get(column_slot.as_usize() % T::EthSpec::slots_per_epoch() as usize)
-            .ok_or_else(|| BeaconChainError::NoProposerForSlot(column_slot))?;
-
         // Prime the proposer shuffling cache with the newly-learned value.
-        chain.beacon_proposer_cache.lock().insert(
-            column_epoch,
-            proposer_shuffling_root,
-            proposers,
-            state.fork(),
-        )?;
-        (proposer_index, state.fork())
-    };
+        Ok::<_, GossipDataColumnError>(EpochBlockProposers {
+            epoch: column_epoch,
+            fork: state.fork(),
+            proposers: proposers.into(),
+        })
+    })?;
+
+    let proposer_index = *epoch_proposers
+        .proposers
+        .get(column_slot.as_usize() % T::EthSpec::slots_per_epoch() as usize)
+        .ok_or_else(|| BeaconChainError::NoProposerForSlot(column_slot))?;
+
+    let fork = epoch_proposers.fork;
 
     // Signature verify the signed block header.
     let signature_is_valid = {

common/logging/Cargo.toml

@@ -11,8 +11,6 @@ test_logger = [] # Print log output to stderr when running tests instead of drop
 chrono = { version = "0.4", default-features = false, features = ["clock", "std"] }
 logroller = { workspace = true }
 metrics = { workspace = true }
-once_cell = "1.17.1"
-parking_lot = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = [ "time" ] }

testing/simulator/src/basic_sim.rs

@@ -18,17 +18,22 @@ use environment::tracing_common;
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
+use logging::build_workspace_filter;
 use tokio::time::sleep;
 use types::{Epoch, EthSpec, MinimalEthSpec};
 
 const END_EPOCH: u64 = 16;
 const GENESIS_DELAY: u64 = 32;
 const ALTAIR_FORK_EPOCH: u64 = 0;
 const BELLATRIX_FORK_EPOCH: u64 = 0;
-const CAPELLA_FORK_EPOCH: u64 = 1;
-const DENEB_FORK_EPOCH: u64 = 2;
-// const ELECTRA_FORK_EPOCH: u64 = 3;
-// const FULU_FORK_EPOCH: u64  = 4;
+const CAPELLA_FORK_EPOCH: u64 = 0;
+const DENEB_FORK_EPOCH: u64 = 0;
+const ELECTRA_FORK_EPOCH: u64 = 1;
+// Set Fulu fork epoch half way through the test, so that:
+// 1. We run the simulator for a few electra epochs to cover electra testing;
+// 2. The Fulu fork epoch is not too close to Electra fork epoch, so we're not subscribed to topics
+//    across 3 forks simultaneously.
+const FULU_FORK_EPOCH: u64 = 8;
 
 const SUGGESTED_FEE_RECIPIENT: [u8; 20] =
     [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
@@ -116,7 +121,11 @@ pub fn run_basic_sim(matches: &ArgMatches) -> Result<(), String> {
     );
 
     if let Err(e) = tracing_subscriber::registry()
-        .with(stdout_logging_layer.with_filter(logger_config.debug_level))
+        .with(
+            stdout_logging_layer
+                .with_filter(logger_config.debug_level)
+                .with_filter(build_workspace_filter()?),
+        )
         .try_init()
     {
         eprintln!("Failed to initialize dependency logging: {e}");
@@ -130,8 +139,8 @@ pub fn run_basic_sim(matches: &ArgMatches) -> Result<(), String> {
     let genesis_delay = GENESIS_DELAY;
 
     // Convenience variables. Update these values when adding a newer fork.
-    let latest_fork_version = spec.deneb_fork_version;
-    let latest_fork_start_epoch = DENEB_FORK_EPOCH;
+    let latest_fork_version = spec.fulu_fork_version;
+    let latest_fork_start_epoch = FULU_FORK_EPOCH;
 
     spec.seconds_per_slot /= speed_up_factor;
     spec.seconds_per_slot = max(1, spec.seconds_per_slot);
@@ -142,8 +151,8 @@ pub fn run_basic_sim(matches: &ArgMatches) -> Result<(), String> {
     spec.bellatrix_fork_epoch = Some(Epoch::new(BELLATRIX_FORK_EPOCH));
     spec.capella_fork_epoch = Some(Epoch::new(CAPELLA_FORK_EPOCH));
     spec.deneb_fork_epoch = Some(Epoch::new(DENEB_FORK_EPOCH));
-    //spec.electra_fork_epoch = Some(Epoch::new(ELECTRA_FORK_EPOCH));
-    //spec.fulu_fork_epoch = Some(Epoch::new(FULU_FORK_EPOCH));
+    spec.electra_fork_epoch = Some(Epoch::new(ELECTRA_FORK_EPOCH));
+    spec.fulu_fork_epoch = Some(Epoch::new(FULU_FORK_EPOCH));
     let spec = Arc::new(spec);
     env.eth2_config.spec = spec.clone();
 
@@ -174,6 +183,13 @@ pub fn run_basic_sim(matches: &ArgMatches) -> Result<(), String> {
         })
         .await?;
 
+        // Run extra node as a PeerDAS full node to cover full node code path and syncing.
+        let extra_node_beacon_config = {
+            let mut config = beacon_config.clone();
+            config.network.subscribe_all_data_column_subnets = true;
+            config
+        };
+
         // Add nodes to the network.
         for _ in 0..node_count {
             network
@@ -319,7 +335,7 @@ pub fn run_basic_sim(matches: &ArgMatches) -> Result<(), String> {
                 slot_duration
             ),
             network_1.add_beacon_node_with_delay(
-                beacon_config.clone(),
+                extra_node_beacon_config,
                 mock_execution_config.clone(),
                 END_EPOCH - 1,
                 slot_duration,

testing/simulator/src/checks.rs

@@ -128,17 +128,23 @@ pub async fn verify_full_block_production_up_to<E: EthSpec>(
     slot_delay(slot, slot_duration).await;
     let beacon_nodes = network.beacon_nodes.read();
     let beacon_chain = beacon_nodes[0].client.beacon_chain().unwrap();
-    let num_blocks = beacon_chain
+    let block_slots = beacon_chain
         .chain_dump()
         .unwrap()
         .iter()
         .take_while(|s| s.beacon_block.slot() <= slot)
-        .count();
+        .map(|s| s.beacon_block.slot().as_usize())
+        .collect::<Vec<_>>();
+    let num_blocks = block_slots.len();
     if num_blocks != slot.as_usize() + 1 {
+        let missed_slots = (0..slot.as_usize())
+            .filter(|slot| !block_slots.contains(slot))
+            .collect::<Vec<_>>();
         return Err(format!(
-            "There wasn't a block produced at every slot, got: {}, expected: {}",
+            "There wasn't a block produced at every slot, got: {}, expected: {}, missed: {:?}",
             num_blocks,
-            slot.as_usize() + 1
+            slot.as_usize() + 1,
+            missed_slots
         ));
     }
     Ok(())
@@ -185,12 +191,17 @@ pub async fn verify_full_sync_aggregates_up_to<E: EthSpec>(
             .get_beacon_blocks::<E>(BlockId::Slot(Slot::new(slot)))
             .await
             .map(|resp| {
-                resp.unwrap()
-                    .data
-                    .message()
-                    .body()
-                    .sync_aggregate()
-                    .map(|agg| agg.num_set_bits())
+                resp.unwrap_or_else(|| {
+                    panic!(
+                        "Beacon block for slot {} not returned from Beacon API",
+                        slot
+                    )
+                })
+                .data
+                .message()
+                .body()
+                .sync_aggregate()
+                .map(|agg| agg.num_set_bits())
             })
             .map_err(|e| format!("Error while getting beacon block: {:?}", e))?
             .map_err(|_| format!("Altair block {} should have sync aggregate", slot))?;

testing/simulator/src/local_network.rs

@@ -43,6 +43,7 @@ fn default_client_config(network_params: LocalNetworkParams, genesis_time: u64)
     beacon_config.network.enr_address = (Some(Ipv4Addr::LOCALHOST), None);
     beacon_config.network.enable_light_client_server = true;
     beacon_config.network.discv5_config.enable_packet_filter = false;
+    beacon_config.network.subscribe_all_data_column_subnets = true;
     beacon_config.chain.enable_light_client_server = true;
     beacon_config.chain.optimistic_finalized_sync = false;
     beacon_config.trusted_setup = serde_json::from_reader(get_trusted_setup().as_slice())