Upgrade in-memory and on-disk state representation with tree states

Cargo.toml

@@ -37,6 +37,7 @@ members = [
     "common/malloc_utils",
     "common/oneshot_broadcast",
     "common/pretty_reqwest_error",
+    "common/promise_cache",
     "common/sensitive_url",
     "common/slot_clock",
     "common/system_health",
@@ -102,6 +103,7 @@ bytes = "1"
 clap = "2"
 compare_fields_derive = { path = "common/compare_fields_derive" }
 criterion = "0.3"
+crossbeam-channel = "0.5.8"
 delay_map = "0.3"
 derivative = "2"
 dirs = "3"
@@ -127,6 +129,7 @@ libsecp256k1 = "0.7"
 log = "0.4"
 lru = "0.12"
 maplit = "1"
+milhouse = { git = "https://github.com/sigp/milhouse", branch = "main" }
 num_cpus = "1"
 parking_lot = "0.12"
 paste = "1"
@@ -153,7 +156,7 @@ smallvec = "1.11.2"
 snap = "1"
 ssz_types = "0.5"
 strum = { version = "0.24", features = ["derive"] }
-superstruct = "0.6"
+superstruct = "0.7"
 syn = "1"
 sysinfo = "0.26"
 tempfile = "3"
@@ -170,8 +173,10 @@ url = "2"
 uuid = { version = "0.8", features = ["serde", "v4"] }
 # TODO update to warp 0.3.6 after released.
 warp = { git = "https://github.com/seanmonstar/warp.git", default-features = false, features = ["tls"] }
+xdelta3 = { git = "http://github.com/michaelsproul/xdelta3-rs", rev="ae9a1d2585ef998f4656acdc35cf263ee88e6dfa" }
 zeroize = { version = "1", features = ["zeroize_derive"] }
 zip = "0.6"
+zstd = "0.11.2"
 
 # Local crates.
 account_utils = { path = "common/account_utils" }

account_manager/src/validator/exit.rs

@@ -203,8 +203,8 @@ async fn publish_voluntary_exit<E: EthSpec>(
         let validator_data = get_validator_data(client, &keypair.pk).await?;
         match validator_data.status {
             ValidatorStatus::ActiveExiting => {
-                let exit_epoch = validator_data.validator.exit_epoch;
-                let withdrawal_epoch = validator_data.validator.withdrawable_epoch;
+                let exit_epoch = validator_data.validator.exit_epoch();
+                let withdrawal_epoch = validator_data.validator.withdrawable_epoch();
                 let current_epoch = get_current_epoch::<E>(genesis_data.genesis_time, spec)
                     .ok_or("Failed to get current epoch. Please check your system time")?;
                 eprintln!("Voluntary exit has been accepted into the beacon chain, but not yet finalized. \
@@ -224,7 +224,7 @@ async fn publish_voluntary_exit<E: EthSpec>(
             ValidatorStatus::ExitedSlashed | ValidatorStatus::ExitedUnslashed => {
                 eprintln!(
                     "Validator has exited on epoch: {}",
-                    validator_data.validator.exit_epoch
+                    validator_data.validator.exit_epoch()
                 );
                 break;
             }
@@ -250,7 +250,7 @@ async fn get_validator_index_for_exit(
         ValidatorStatus::ActiveOngoing => {
             let eligible_epoch = validator_data
                 .validator
-                .activation_epoch
+                .activation_epoch()
                 .safe_add(spec.shard_committee_period)
                 .map_err(|e| format!("Failed to calculate eligible epoch, validator activation epoch too high: {:?}", e))?;
 

beacon_node/Cargo.toml

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

beacon_node/beacon_chain/Cargo.toml

@@ -18,59 +18,60 @@ environment = { workspace = true }
 serde_json = { workspace = true }
 
 [dependencies]
-serde_json = { workspace = true }
+bitvec = { workspace = true }
+bls = { workspace = true }
+crossbeam-channel = { workspace = true }
+derivative = { workspace = true }
+eth1 = { workspace = true }
+eth2 = { workspace = true }
 eth2_network_config = { workspace = true }
-merkle_proof = { workspace = true }
-store = { workspace = true }
-parking_lot = { workspace = true }
+ethereum_hashing = { workspace = true }
+ethereum_serde_utils = { workspace = true }
+ethereum_ssz = { workspace = true }
+ethereum_ssz_derive = { workspace = true }
+execution_layer = { workspace = true }
+exit-future = { workspace = true }
+fork_choice = { workspace = true }
+futures = { workspace = true }
+genesis = { workspace = true }
+hex = { workspace = true }
+int_to_bytes = { workspace = true }
+itertools = { workspace = true }
+kzg = { workspace = true }
 lazy_static = { workspace = true }
-smallvec = { workspace = true }
 lighthouse_metrics = { workspace = true }
+logging = { workspace = true }
+lru = { workspace = true }
+merkle_proof = { workspace = true }
 operation_pool = { workspace = true }
+parking_lot = { workspace = true }
+promise_cache = { path = "../../common/promise_cache" }
+proto_array = { workspace = true }
+rand = { workspace = true }
 rayon = { workspace = true }
+safe_arith = { workspace = true }
+sensitive_url = { workspace = true }
 serde = { workspace = true }
-ethereum_serde_utils = { workspace = true }
+serde_json = { workspace = true }
+slasher = { workspace = true }
 slog = { workspace = true }
+slog-async = { workspace = true }
+slog-term = { workspace = true }
 sloggers = { workspace = true }
 slot_clock = { workspace = true }
-ethereum_hashing = { workspace = true }
-ethereum_ssz = { workspace = true }
+smallvec = { workspace = true }
 ssz_types = { workspace = true }
-ethereum_ssz_derive = { workspace = true }
 state_processing = { workspace = true }
-tree_hash_derive = { workspace = true }
-tree_hash = { workspace = true }
-types = { workspace = true }
-tokio = { workspace = true }
-tokio-stream = { workspace = true }
-eth1 = { workspace = true }
-futures = { workspace = true }
-genesis = { workspace = true }
-int_to_bytes = { workspace = true }
-rand = { workspace = true }
-proto_array = { workspace = true }
-lru = { workspace = true }
-tempfile = { workspace = true }
-bitvec = { workspace = true }
-bls = { workspace = true }
-kzg = { workspace = true }
-safe_arith = { workspace = true }
-fork_choice = { workspace = true }
-task_executor = { workspace = true }
-derivative = { workspace = true }
-itertools = { workspace = true }
-slasher = { workspace = true }
-eth2 = { workspace = true }
+store = { workspace = true }
 strum = { workspace = true }
-logging = { workspace = true }
-execution_layer = { workspace = true }
-sensitive_url = { workspace = true }
 superstruct = { workspace = true }
-hex = { workspace = true }
-exit-future = { workspace = true }
-oneshot_broadcast = { path = "../../common/oneshot_broadcast/" }
-slog-term = { workspace = true }
-slog-async = { workspace = true }
+task_executor = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
+tree_hash = { workspace = true }
+tree_hash_derive = { workspace = true }
+types = { workspace = true }
 
 [[test]]
 name = "beacon_chain_tests"

beacon_node/beacon_chain/src/attestation_rewards.rs

@@ -6,7 +6,7 @@ use safe_arith::SafeArith;
 use serde_utils::quoted_u64::Quoted;
 use slog::debug;
 use state_processing::per_epoch_processing::altair::{
-    process_inactivity_updates, process_justification_and_finalization,
+    process_inactivity_updates_slow, process_justification_and_finalization,
 };
 use state_processing::{
     common::altair::BaseRewardPerIncrement,
@@ -134,10 +134,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let spec = &self.spec;
 
         // Calculate ideal_rewards
-        let participation_cache = ParticipationCache::new(&state, spec)?;
-        process_justification_and_finalization(&state, &participation_cache)?
-            .apply_changes_to_state(&mut state);
-        process_inactivity_updates(&mut state, &participation_cache, spec)?;
+        let participation_cache = ParticipationCache::new(&state, spec)
+            .map_err(|_| BeaconChainError::AttestationRewardsError)?;
+        process_justification_and_finalization(&state)?.apply_changes_to_state(&mut state);
+        process_inactivity_updates_slow(&mut state, spec)?;
 
         let previous_epoch = state.previous_epoch();
 
@@ -147,13 +147,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             let weight = get_flag_weight(flag_index)
                 .map_err(|_| BeaconChainError::AttestationRewardsError)?;
 
-            let unslashed_participating_indices = participation_cache
-                .get_unslashed_participating_indices(flag_index, previous_epoch)?;
-
-            let unslashed_participating_balance =
-                unslashed_participating_indices
-                    .total_balance()
-                    .map_err(|_| BeaconChainError::AttestationRewardsError)?;
+            let unslashed_participating_balance = participation_cache
+                .previous_epoch_flag_attesting_balance(flag_index)
+                .map_err(|_| BeaconChainError::AttestationRewardsError)?;
 
             let unslashed_participating_increments =
                 unslashed_participating_balance.safe_div(spec.effective_balance_increment)?;
@@ -199,24 +195,41 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             Self::validators_ids_to_indices(&mut state, validators)?
         };
 
-        for validator_index in &validators {
-            let eligible = state.is_eligible_validator(previous_epoch, *validator_index)?;
+        for &validator_index in &validators {
+            // Return 0s for unknown/inactive validator indices. This is a bit different from stable
+            // where we error for unknown pubkeys.
+            let Ok(validator) = participation_cache.get_validator(validator_index) else {
+                debug!(
+                    self.log,
+                    "No rewards for inactive/unknown validator";
+                    "index" => validator_index,
+                    "epoch" => previous_epoch
+                );
+                total_rewards.push(TotalAttestationRewards {
+                    validator_index: validator_index as u64,
+                    head: 0,
+                    target: 0,
+                    source: 0,
+                    inclusion_delay: None,
+                    inactivity: 0,
+                });
+                continue;
+            };
+            let eligible = validator.is_eligible;
             let mut head_reward = 0i64;
             let mut target_reward = 0i64;
             let mut source_reward = 0i64;
             let mut inactivity_penalty = 0i64;
 
             if eligible {
-                let effective_balance = state.get_effective_balance(*validator_index)?;
+                let effective_balance = validator.effective_balance;
 
                 for flag_index in 0..PARTICIPATION_FLAG_WEIGHTS.len() {
                     let (ideal_reward, penalty) = ideal_rewards_hashmap
                         .get(&(flag_index, effective_balance))
                         .ok_or(BeaconChainError::AttestationRewardsError)?;
-                    let voted_correctly = participation_cache
-                        .get_unslashed_participating_indices(flag_index, previous_epoch)
-                        .map_err(|_| BeaconChainError::AttestationRewardsError)?
-                        .contains(*validator_index)
+                    let voted_correctly = validator
+                        .is_unslashed_participating_index(flag_index)
                         .map_err(|_| BeaconChainError::AttestationRewardsError)?;
                     if voted_correctly {
                         if flag_index == TIMELY_HEAD_FLAG_INDEX {
@@ -232,7 +245,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         target_reward = *penalty;
 
                         let penalty_numerator = effective_balance
-                            .safe_mul(state.get_inactivity_score(*validator_index)?)?;
+                            .safe_mul(state.get_inactivity_score(validator_index)?)?;
                         let penalty_denominator = spec
                             .inactivity_score_bias
                             .safe_mul(spec.inactivity_penalty_quotient_for_state(&state))?;
@@ -244,7 +257,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 }
             }
             total_rewards.push(TotalAttestationRewards {
-                validator_index: *validator_index as u64,
+                validator_index: validator_index as u64,
                 head: head_reward,
                 target: target_reward,
                 source: source_reward,

beacon_node/beacon_chain/src/beacon_block_reward.rs

@@ -4,9 +4,8 @@ use operation_pool::RewardCache;
 use safe_arith::SafeArith;
 use slog::error;
 use state_processing::{
-    common::{
-        altair, get_attestation_participation_flag_indices, get_attesting_indices_from_state,
-    },
+    common::{get_attestation_participation_flag_indices, get_attesting_indices_from_state},
+    epoch_cache::initialize_epoch_cache,
     per_block_processing::{
         altair::sync_committee::compute_sync_aggregate_rewards, get_slashable_indices,
     },
@@ -32,6 +31,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         state.build_committee_cache(RelativeEpoch::Previous, &self.spec)?;
         state.build_committee_cache(RelativeEpoch::Current, &self.spec)?;
+        initialize_epoch_cache(state, &self.spec)?;
 
         self.compute_beacon_block_reward_with_cache(block, block_root, state)
     }
@@ -135,7 +135,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             proposer_slashing_reward.safe_add_assign(
                 state
                     .get_validator(proposer_slashing.proposer_index() as usize)?
-                    .effective_balance
+                    .effective_balance()
                     .safe_div(self.spec.whistleblower_reward_quotient)?,
             )?;
         }
@@ -157,7 +157,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 attester_slashing_reward.safe_add_assign(
                     state
                         .get_validator(attester_index as usize)?
-                        .effective_balance
+                        .effective_balance()
                         .safe_div(self.spec.whistleblower_reward_quotient)?,
                 )?;
             }
@@ -191,10 +191,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         block: BeaconBlockRef<'_, T::EthSpec, Payload>,
         state: &BeaconState<T::EthSpec>,
     ) -> Result<BeaconBlockSubRewardValue, BeaconChainError> {
-        let total_active_balance = state.get_total_active_balance()?;
-        let base_reward_per_increment =
-            altair::BaseRewardPerIncrement::new(total_active_balance, &self.spec)?;
-
         let mut total_proposer_reward = 0;
 
         let proposer_reward_denominator = WEIGHT_DENOMINATOR
@@ -235,15 +231,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         && !validator_participation.has_flag(flag_index)?
                     {
                         validator_participation.add_flag(flag_index)?;
-                        proposer_reward_numerator.safe_add_assign(
-                            altair::get_base_reward(
-                                state,
-                                index,
-                                base_reward_per_increment,
-                                &self.spec,
-                            )?
-                            .safe_mul(weight)?,
-                        )?;
+                        proposer_reward_numerator
+                            .safe_add_assign(state.get_base_reward(index)?.safe_mul(weight)?)?;
                     }
                 }
             }

beacon_node/beacon_chain/src/beacon_block_streamer.rs

@@ -430,7 +430,7 @@ impl<T: BeaconChainTypes> BeaconBlockStreamer<T> {
                 continue;
             }
 
-            match self.beacon_chain.store.try_get_full_block(&root) {
+            match self.beacon_chain.store.try_get_full_block(&root, None) {
                 Err(e) => db_blocks.push((root, Err(e.into()))),
                 Ok(opt_block) => db_blocks.push((
                     root,