From 08e045875f312f56356553332e6c43bc629e1c8a Mon Sep 17 00:00:00 2001 From: Michael Sproul Date: Tue, 7 May 2024 16:20:50 +1000 Subject: [PATCH 1/2] Electra epoch processing --- consensus/state_processing/src/epoch_cache.rs | 42 ++- .../src/per_epoch_processing/errors.rs | 1 + .../src/per_epoch_processing/single_pass.rs | 338 ++++++++++++++++-- consensus/types/src/validator.rs | 30 +- 4 files changed, 369 insertions(+), 42 deletions(-) diff --git a/consensus/state_processing/src/epoch_cache.rs b/consensus/state_processing/src/epoch_cache.rs index b2f2d85407..3741f49c57 100644 --- a/consensus/state_processing/src/epoch_cache.rs +++ b/consensus/state_processing/src/epoch_cache.rs @@ -9,6 +9,7 @@ use types::{ActivationQueue, BeaconState, ChainSpec, EthSpec, ForkName, Hash256} pub struct PreEpochCache { epoch_key: EpochCacheKey, effective_balances: Vec, + total_active_balance: u64, } impl PreEpochCache { @@ -36,20 +37,51 @@ impl PreEpochCache { Ok(Self { epoch_key, effective_balances: Vec::with_capacity(state.validators().len()), + total_active_balance: 0, }) } - pub fn push_effective_balance(&mut self, effective_balance: u64) { - self.effective_balances.push(effective_balance); + pub fn update_effective_balance( + &mut self, + validator_index: usize, + effective_balance: u64, + is_active_next_epoch: bool, + ) -> Result<(), EpochCacheError> { + if validator_index == self.effective_balances.len() { + self.effective_balances.push(effective_balance); + if is_active_next_epoch { + self.total_active_balance + .safe_add_assign(effective_balance)?; + } + + Ok(()) + } else if let Some(existing_balance) = self.effective_balances.get_mut(validator_index) { + // Update total active balance for a late change in effective balance. This happens when + // processing consolidations. + if is_active_next_epoch { + self.total_active_balance + .safe_add_assign(effective_balance)?; + self.total_active_balance + .safe_sub_assign(*existing_balance)?; + } + *existing_balance = effective_balance; + Ok(()) + } else { + Err(EpochCacheError::ValidatorIndexOutOfBounds { validator_index }) + } + } + + pub fn get_total_active_balance(&self) -> u64 { + self.total_active_balance } pub fn into_epoch_cache( self, - total_active_balance: u64, activation_queue: ActivationQueue, spec: &ChainSpec, ) -> Result { let epoch = self.epoch_key.epoch; + let total_active_balance = self.total_active_balance; let sqrt_total_active_balance = SqrtTotalActiveBalance::new(total_active_balance); let base_reward_per_increment = BaseRewardPerIncrement::new(total_active_balance, spec)?; @@ -131,9 +163,9 @@ pub fn initialize_epoch_cache( decision_block_root, }, effective_balances, + total_active_balance, }; - *state.epoch_cache_mut() = - pre_epoch_cache.into_epoch_cache(total_active_balance, activation_queue, spec)?; + *state.epoch_cache_mut() = pre_epoch_cache.into_epoch_cache(activation_queue, spec)?; Ok(()) } diff --git a/consensus/state_processing/src/per_epoch_processing/errors.rs b/consensus/state_processing/src/per_epoch_processing/errors.rs index de481ec676..7f60f4a6e0 100644 --- a/consensus/state_processing/src/per_epoch_processing/errors.rs +++ b/consensus/state_processing/src/per_epoch_processing/errors.rs @@ -25,6 +25,7 @@ pub enum EpochProcessingError { InvalidFlagIndex(usize), MilhouseError(milhouse::Error), EpochCache(EpochCacheError), + SinglePassMissingActivationQueue, } impl From for EpochProcessingError { diff --git a/consensus/state_processing/src/per_epoch_processing/single_pass.rs b/consensus/state_processing/src/per_epoch_processing/single_pass.rs index a9629e73e4..3d72f0cdd8 100644 --- a/consensus/state_processing/src/per_epoch_processing/single_pass.rs +++ b/consensus/state_processing/src/per_epoch_processing/single_pass.rs @@ -1,20 +1,24 @@ use crate::{ - common::update_progressive_balances_cache::initialize_progressive_balances_cache, + common::{ + decrease_balance, increase_balance, + update_progressive_balances_cache::initialize_progressive_balances_cache, + }, epoch_cache::{initialize_epoch_cache, PreEpochCache}, per_epoch_processing::{Delta, Error, ParticipationEpochSummary}, }; use itertools::izip; use safe_arith::{SafeArith, SafeArithIter}; use std::cmp::{max, min}; -use std::collections::BTreeSet; +use std::collections::{BTreeSet, HashMap}; use types::{ consts::altair::{ NUM_FLAG_INDICES, PARTICIPATION_FLAG_WEIGHTS, TIMELY_HEAD_FLAG_INDEX, TIMELY_TARGET_FLAG_INDEX, WEIGHT_DENOMINATOR, }, milhouse::Cow, - ActivationQueue, BeaconState, BeaconStateError, ChainSpec, Epoch, EthSpec, ExitCache, ForkName, - ParticipationFlags, ProgressiveBalancesCache, RelativeEpoch, Unsigned, Validator, + ActivationQueue, BeaconState, BeaconStateError, ChainSpec, Checkpoint, Epoch, EthSpec, + ExitCache, ForkName, List, ParticipationFlags, ProgressiveBalancesCache, RelativeEpoch, + Unsigned, Validator, }; pub struct SinglePassConfig { @@ -22,6 +26,8 @@ pub struct SinglePassConfig { pub rewards_and_penalties: bool, pub registry_updates: bool, pub slashings: bool, + pub pending_balance_deposits: bool, + pub pending_consolidations: bool, pub effective_balance_updates: bool, } @@ -38,6 +44,8 @@ impl SinglePassConfig { rewards_and_penalties: true, registry_updates: true, slashings: true, + pending_balance_deposits: true, + pending_consolidations: true, effective_balance_updates: true, } } @@ -48,6 +56,8 @@ impl SinglePassConfig { rewards_and_penalties: false, registry_updates: false, slashings: false, + pending_balance_deposits: false, + pending_consolidations: false, effective_balance_updates: false, } } @@ -57,6 +67,7 @@ impl SinglePassConfig { struct StateContext { current_epoch: Epoch, next_epoch: Epoch, + finalized_checkpoint: Checkpoint, is_in_inactivity_leak: bool, total_active_balance: u64, churn_limit: u64, @@ -73,6 +84,15 @@ struct SlashingsContext { target_withdrawable_epoch: Epoch, } +struct PendingBalanceDepositsContext { + /// The value to set `next_deposit_index` to *after* processing completes. + next_deposit_index: usize, + /// The value to set `deposit_balance_to_consume` to *after* processing completes. + deposit_balance_to_consume: u64, + /// Total balance increases for each validator due to pending balance deposits. + validator_deposits_to_process: HashMap, +} + struct EffectiveBalancesContext { downward_threshold: u64, upward_threshold: u64, @@ -129,6 +149,7 @@ pub fn process_epoch_single_pass( let state_ctxt = &StateContext { current_epoch, next_epoch, + finalized_checkpoint, is_in_inactivity_leak, total_active_balance, churn_limit, @@ -139,6 +160,13 @@ pub fn process_epoch_single_pass( let slashings_ctxt = &SlashingsContext::new(state, state_ctxt, spec)?; let mut next_epoch_cache = PreEpochCache::new_for_next_epoch(state)?; + let pending_balance_deposits_ctxt = + if fork_name >= ForkName::Electra && conf.pending_balance_deposits { + Some(PendingBalanceDepositsContext::new(state, spec)?) + } else { + None + }; + // Split the state into several disjoint mutable borrows. let ( validators, @@ -165,12 +193,19 @@ pub fn process_epoch_single_pass( // Compute shared values required for different parts of epoch processing. let rewards_ctxt = &RewardsAndPenaltiesContext::new(progressive_balances, state_ctxt, spec)?; - let activation_queue = &epoch_cache - .activation_queue()? - .get_validators_eligible_for_activation( - finalized_checkpoint.epoch, - activation_churn_limit as usize, - ); + + let mut activation_queues = if fork_name < ForkName::Electra { + let activation_queue = epoch_cache + .activation_queue()? + .get_validators_eligible_for_activation( + finalized_checkpoint.epoch, + activation_churn_limit as usize, + ); + let next_epoch_activation_queue = ActivationQueue::default(); + Some((activation_queue, next_epoch_activation_queue)) + } else { + None + }; let effective_balances_ctxt = &EffectiveBalancesContext::new(spec)?; // Iterate over the validators and related fields in one pass. @@ -178,10 +213,6 @@ pub fn process_epoch_single_pass( let mut balances_iter = balances.iter_cow(); let mut inactivity_scores_iter = inactivity_scores.iter_cow(); - // Values computed for the next epoch transition. - let mut next_epoch_total_active_balance = 0; - let mut next_epoch_activation_queue = ActivationQueue::default(); - for (index, &previous_epoch_participation, ¤t_epoch_participation) in izip!( 0..num_validators, previous_epoch_participation.iter(), @@ -246,12 +277,14 @@ pub fn process_epoch_single_pass( // `process_registry_updates` if conf.registry_updates { + let activation_queue_refs = activation_queues + .as_mut() + .map(|(current_queue, next_queue)| (&*current_queue, next_queue)); process_single_registry_update( &mut validator, validator_info, exit_cache, - activation_queue, - &mut next_epoch_activation_queue, + activation_queue_refs, state_ctxt, spec, )?; @@ -262,13 +295,22 @@ pub fn process_epoch_single_pass( process_single_slashing(&mut balance, &validator, slashings_ctxt, state_ctxt, spec)?; } + // `process_pending_balance_deposits` + if let Some(pending_balance_deposits_ctxt) = &pending_balance_deposits_ctxt { + process_pending_balance_deposits_for_validator( + &mut balance, + validator_info, + pending_balance_deposits_ctxt, + )?; + } + // `process_effective_balance_updates` if conf.effective_balance_updates { process_single_effective_balance_update( + validator_info.index, *balance, &mut validator, - validator_info, - &mut next_epoch_total_active_balance, + validator_info.current_epoch_participation, &mut next_epoch_cache, progressive_balances, effective_balances_ctxt, @@ -278,15 +320,45 @@ pub fn process_epoch_single_pass( } } - if conf.effective_balance_updates { - state.set_total_active_balance(next_epoch, next_epoch_total_active_balance, spec); - *state.epoch_cache_mut() = next_epoch_cache.into_epoch_cache( - next_epoch_total_active_balance, - next_epoch_activation_queue, + // Finish processing pending balance deposits if relevant. + // + // This *could* be reordered after `process_pending_consolidations` which pushes only to the end + // of the `pending_balance_deposits` list. But we may as well preserve the write ordering used + // by the spec and do this first. + if let Some(ctxt) = pending_balance_deposits_ctxt { + let new_pending_balance_deposits = List::try_from_iter( + state + .pending_balance_deposits()? + .iter_from(ctxt.next_deposit_index)? + .cloned(), + )?; + *state.pending_balance_deposits_mut()? = new_pending_balance_deposits; + *state.deposit_balance_to_consume_mut()? = ctxt.deposit_balance_to_consume; + } + + // Process consolidations outside the single-pass loop, as they depend on balances for multiple + // validators and cannot be computed accurately inside the loop. + if fork_name >= ForkName::Electra && conf.pending_consolidations { + process_pending_consolidations( + state, + &mut next_epoch_cache, + effective_balances_ctxt, + state_ctxt, spec, )?; } + // Finally, finish updating effective balance caches. We need this to happen *after* processing + // of pending consolidations, which recomputes some effective balances. + if conf.effective_balance_updates { + let next_epoch_total_active_balance = next_epoch_cache.get_total_active_balance(); + state.set_total_active_balance(next_epoch, next_epoch_total_active_balance, spec); + let next_epoch_activation_queue = + activation_queues.map_or_else(ActivationQueue::default, |(_, queue)| queue); + *state.epoch_cache_mut() = + next_epoch_cache.into_epoch_cache(next_epoch_activation_queue, spec)?; + } + Ok(summary) } @@ -456,6 +528,31 @@ impl RewardsAndPenaltiesContext { } fn process_single_registry_update( + validator: &mut Cow, + validator_info: &ValidatorInfo, + exit_cache: &mut ExitCache, + activation_queues: Option<(&BTreeSet, &mut ActivationQueue)>, + state_ctxt: &StateContext, + spec: &ChainSpec, +) -> Result<(), Error> { + if state_ctxt.fork_name < ForkName::Electra { + let (activation_queue, next_epoch_activation_queue) = + activation_queues.ok_or(Error::SinglePassMissingActivationQueue)?; + process_single_registry_update_pre_electra( + validator, + validator_info, + exit_cache, + activation_queue, + next_epoch_activation_queue, + state_ctxt, + spec, + ) + } else { + process_single_registry_update_post_electra(validator, exit_cache, state_ctxt, spec) + } +} + +fn process_single_registry_update_pre_electra( validator: &mut Cow, validator_info: &ValidatorInfo, exit_cache: &mut ExitCache, @@ -491,6 +588,35 @@ fn process_single_registry_update( Ok(()) } +fn process_single_registry_update_post_electra( + validator: &mut Cow, + exit_cache: &mut ExitCache, + state_ctxt: &StateContext, + spec: &ChainSpec, +) -> Result<(), Error> { + let current_epoch = state_ctxt.current_epoch; + + if validator.is_eligible_for_activation_queue(spec, state_ctxt.fork_name) { + validator.make_mut()?.activation_eligibility_epoch = current_epoch.safe_add(1)?; + } + + if validator.is_active_at(current_epoch) && validator.effective_balance <= spec.ejection_balance + { + // TODO(electra): make sure initiate_validator_exit is updated + initiate_validator_exit(validator, exit_cache, state_ctxt, spec)?; + } + + if validator.is_eligible_for_activation_with_finalized_checkpoint( + &state_ctxt.finalized_checkpoint, + spec, + ) { + validator.make_mut()?.activation_epoch = + spec.compute_activation_exit_epoch(current_epoch)?; + } + + Ok(()) +} + fn initiate_validator_exit( validator: &mut Cow, exit_cache: &mut ExitCache, @@ -568,6 +694,146 @@ fn process_single_slashing( Ok(()) } +impl PendingBalanceDepositsContext { + fn new(state: &BeaconState, spec: &ChainSpec) -> Result { + let available_for_processing = state + .deposit_balance_to_consume()? + .safe_add(state.get_activation_exit_churn_limit(spec)?)?; + let mut processed_amount = 0; + let mut next_deposit_index = 0; + let mut validator_deposits_to_process = HashMap::new(); + + let pending_balance_deposits = state.pending_balance_deposits()?; + + for deposit in pending_balance_deposits.iter() { + if processed_amount.safe_add(deposit.amount)? > available_for_processing { + break; + } + validator_deposits_to_process + .entry(deposit.index as usize) + .or_insert(0) + .safe_add_assign(deposit.amount)?; + + processed_amount.safe_add_assign(deposit.amount)?; + next_deposit_index.safe_add_assign(1)?; + } + + let deposit_balance_to_consume = if next_deposit_index == pending_balance_deposits.len() { + 0 + } else { + available_for_processing.safe_sub(processed_amount)? + }; + + Ok(Self { + next_deposit_index, + deposit_balance_to_consume, + validator_deposits_to_process, + }) + } +} + +fn process_pending_balance_deposits_for_validator( + balance: &mut Cow, + validator_info: &ValidatorInfo, + pending_balance_deposits_ctxt: &PendingBalanceDepositsContext, +) -> Result<(), Error> { + if let Some(deposit_amount) = pending_balance_deposits_ctxt + .validator_deposits_to_process + .get(&validator_info.index) + { + balance.make_mut()?.safe_add_assign(*deposit_amount)?; + } + Ok(()) +} + +/// We process pending consolidations after all of single-pass epoch processing, and then patch up +/// the effective balances for affected validators. +/// +/// This is safe because processing consolidations does not depend on the `effective_balance`. +fn process_pending_consolidations( + state: &mut BeaconState, + next_epoch_cache: &mut PreEpochCache, + effective_balances_ctxt: &EffectiveBalancesContext, + state_ctxt: &StateContext, + spec: &ChainSpec, +) -> Result<(), Error> { + let mut next_pending_consolidation: usize = 0; + let current_epoch = state.current_epoch(); + let pending_consolidations = state.pending_consolidations()?.clone(); + + let mut affected_validators = BTreeSet::new(); + + for pending_consolidation in &pending_consolidations { + let source_index = pending_consolidation.source_index as usize; + let target_index = pending_consolidation.target_index as usize; + let source_validator = state.get_validator(source_index)?; + if source_validator.slashed { + next_pending_consolidation.safe_add_assign(1)?; + continue; + } + if source_validator.withdrawable_epoch > current_epoch { + break; + } + + // Calculate the active balance while we have the source validator loaded. This is a safe + // reordering. + let source_balance = *state + .balances() + .get(source_index) + .ok_or(BeaconStateError::UnknownValidator(source_index))?; + let active_balance = + source_validator.get_active_balance(source_balance, spec, state_ctxt.fork_name); + + // Churn any target excess active balance of target and raise its max. + state.switch_to_compounding_validator(target_index, spec)?; + + // Move active balance to target. Excess balance is withdrawable. + decrease_balance(state, source_index, active_balance)?; + increase_balance(state, target_index, active_balance)?; + + affected_validators.insert(source_index); + affected_validators.insert(target_index); + + next_pending_consolidation.safe_add_assign(1)?; + } + + let new_pending_consolidations = List::try_from_iter( + state + .pending_consolidations()? + .iter_from(next_pending_consolidation)? + .cloned(), + )?; + *state.pending_consolidations_mut()? = new_pending_consolidations; + + // Re-process effective balance updates for validators affected by consolidations. + let (validators, balances, _, current_epoch_participation, _, progressive_balances, _, _) = + state.mutable_validator_fields()?; + for validator_index in affected_validators { + let balance = *balances + .get(validator_index) + .ok_or(BeaconStateError::UnknownValidator(validator_index))?; + let mut validator = validators + .get_cow(validator_index) + .ok_or(BeaconStateError::UnknownValidator(validator_index))?; + let validator_current_epoch_participation = *current_epoch_participation + .get(validator_index) + .ok_or(BeaconStateError::UnknownValidator(validator_index))?; + + process_single_effective_balance_update( + validator_index, + balance, + &mut validator, + validator_current_epoch_participation, + next_epoch_cache, + progressive_balances, + effective_balances_ctxt, + state_ctxt, + spec, + )?; + } + Ok(()) +} + impl EffectiveBalancesContext { fn new(spec: &ChainSpec) -> Result { let hysteresis_increment = spec @@ -584,18 +850,24 @@ impl EffectiveBalancesContext { } } +/// This function abstracts over phase0 and Electra effective balance processing. #[allow(clippy::too_many_arguments)] fn process_single_effective_balance_update( + validator_index: usize, balance: u64, validator: &mut Cow, - validator_info: &ValidatorInfo, - next_epoch_total_active_balance: &mut u64, + validator_current_epoch_participation: ParticipationFlags, next_epoch_cache: &mut PreEpochCache, progressive_balances: &mut ProgressiveBalancesCache, eb_ctxt: &EffectiveBalancesContext, state_ctxt: &StateContext, spec: &ChainSpec, ) -> Result<(), Error> { + // Use the higher effective balance limit if post-Electra and compounding withdrawal credentials + // are set. + let effective_balance_limit = + validator.get_validator_max_effective_balance(spec, state_ctxt.fork_name); + let old_effective_balance = validator.effective_balance; let new_effective_balance = if balance.safe_add(eb_ctxt.downward_threshold)? < validator.effective_balance @@ -606,15 +878,13 @@ fn process_single_effective_balance_update( { min( balance.safe_sub(balance.safe_rem(spec.effective_balance_increment)?)?, - spec.max_effective_balance, + effective_balance_limit, ) } else { validator.effective_balance }; - if validator.is_active_at(state_ctxt.next_epoch) { - next_epoch_total_active_balance.safe_add_assign(new_effective_balance)?; - } + let is_active_next_epoch = validator.is_active_at(state_ctxt.next_epoch); if new_effective_balance != old_effective_balance { validator.make_mut()?.effective_balance = new_effective_balance; @@ -623,14 +893,18 @@ fn process_single_effective_balance_update( // previous epoch once the epoch transition completes. progressive_balances.on_effective_balance_change( validator.slashed, - validator_info.current_epoch_participation, + validator_current_epoch_participation, old_effective_balance, new_effective_balance, )?; } - // Caching: update next epoch effective balances. - next_epoch_cache.push_effective_balance(new_effective_balance); + // Caching: update next epoch effective balances and total active balance. + next_epoch_cache.update_effective_balance( + validator_index, + new_effective_balance, + is_active_next_epoch, + )?; Ok(()) } diff --git a/consensus/types/src/validator.rs b/consensus/types/src/validator.rs index 86243083f0..82d0ad311b 100644 --- a/consensus/types/src/validator.rs +++ b/consensus/types/src/validator.rs @@ -1,6 +1,6 @@ use crate::{ - test_utils::TestRandom, Address, BeaconState, ChainSpec, Epoch, EthSpec, ForkName, Hash256, - PublicKeyBytes, + test_utils::TestRandom, Address, BeaconState, ChainSpec, Checkpoint, Epoch, EthSpec, ForkName, + Hash256, PublicKeyBytes, }; use serde::{Deserialize, Serialize}; use ssz_derive::{Decode, Encode}; @@ -87,15 +87,25 @@ impl Validator { } /// Returns `true` if the validator is eligible to be activated. - /// - /// Spec v0.12.1 pub fn is_eligible_for_activation( &self, state: &BeaconState, spec: &ChainSpec, + ) -> bool { + self.is_eligible_for_activation_with_finalized_checkpoint( + &state.finalized_checkpoint(), + spec, + ) + } + + /// Returns `true` if the validator is eligible to be activated. + pub fn is_eligible_for_activation_with_finalized_checkpoint( + &self, + finalized_checkpoint: &Checkpoint, + spec: &ChainSpec, ) -> bool { // Placement in queue is finalized - self.activation_eligibility_epoch <= state.finalized_checkpoint().epoch + self.activation_eligibility_epoch <= finalized_checkpoint.epoch // Has not yet been activated && self.activation_epoch == spec.far_future_epoch } @@ -267,6 +277,16 @@ impl Validator { spec.max_effective_balance } } + + pub fn get_active_balance( + &self, + validator_balance: u64, + spec: &ChainSpec, + current_fork: ForkName, + ) -> u64 { + let max_effective_balance = self.get_validator_max_effective_balance(spec, current_fork); + std::cmp::min(validator_balance, max_effective_balance) + } } impl Default for Validator { From ba02ffc162be74a301a2bcbad70d7efdf3b5784c Mon Sep 17 00:00:00 2001 From: realbigsean Date: Fri, 10 May 2024 09:52:51 -0400 Subject: [PATCH 2/2] add deposit limit for old deposit queue --- beacon_node/beacon_chain/src/eth1_chain.rs | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/beacon_node/beacon_chain/src/eth1_chain.rs b/beacon_node/beacon_chain/src/eth1_chain.rs index 31297244e3..229b3fd525 100644 --- a/beacon_node/beacon_chain/src/eth1_chain.rs +++ b/beacon_node/beacon_chain/src/eth1_chain.rs @@ -546,12 +546,20 @@ impl Eth1ChainBackend for CachingEth1Backend { state.eth1_data().deposit_count }; - match deposit_index.cmp(&deposit_count) { + // [New in Electra:EIP6110] + let deposit_index_limit = + if let Ok(deposit_receipts_start_index) = state.deposit_receipts_start_index() { + std::cmp::min(deposit_count, deposit_receipts_start_index) + } else { + deposit_count + }; + + match deposit_index.cmp(&deposit_index_limit) { Ordering::Greater => Err(Error::DepositIndexTooHigh), Ordering::Equal => Ok(vec![]), Ordering::Less => { let next = deposit_index; - let last = std::cmp::min(deposit_count, next + E::MaxDeposits::to_u64()); + let last = std::cmp::min(deposit_index_limit, next + E::MaxDeposits::to_u64()); self.core .deposits()