Merge branch 'electra-engine-api' of https://github.com/sigp/lighthouse into beacon-api-electra

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<u64>,
+    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) {
+    pub fn update_effective_balance(
-        self.effective_balances.push(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<EpochCache, EpochCacheError> {
         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<E: EthSpec>(
             decision_block_root,
         },
         effective_balances,
+        total_active_balance,
     };
-    *state.epoch_cache_mut() =
+    *state.epoch_cache_mut() = pre_epoch_cache.into_epoch_cache(activation_queue, spec)?;
-        pre_epoch_cache.into_epoch_cache(total_active_balance, activation_queue, spec)?;
 
     Ok(())
 }

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<InclusionError> for EpochProcessingError {

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,
+    ActivationQueue, BeaconState, BeaconStateError, ChainSpec, Checkpoint, Epoch, EthSpec,
-    ParticipationFlags, ProgressiveBalancesCache, RelativeEpoch, Unsigned, Validator,
+    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<usize, u64>,
+}
+
 struct EffectiveBalancesContext {
     downward_threshold: u64,
     upward_threshold: u64,
@@ -129,6 +149,7 @@ pub fn process_epoch_single_pass<E: EthSpec>(
     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<E: EthSpec>(
     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<E: EthSpec>(
 
     // 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()?
+    let mut activation_queues = if fork_name < ForkName::Electra {
-        .get_validators_eligible_for_activation(
+        let activation_queue = epoch_cache
-            finalized_checkpoint.epoch,
+            .activation_queue()?
-            activation_churn_limit as usize,
+            .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<E: EthSpec>(
     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, &current_epoch_participation) in izip!(
         0..num_validators,
         previous_epoch_participation.iter(),
@@ -246,12 +277,14 @@ pub fn process_epoch_single_pass<E: EthSpec>(
 
         // `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,
+                activation_queue_refs,
-                &mut next_epoch_activation_queue,
                 state_ctxt,
                 spec,
             )?;
@@ -262,13 +295,22 @@ pub fn process_epoch_single_pass<E: EthSpec>(
             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,
+                validator_info.current_epoch_participation,
-                &mut next_epoch_total_active_balance,
                 &mut next_epoch_cache,
                 progressive_balances,
                 effective_balances_ctxt,
@@ -278,15 +320,45 @@ pub fn process_epoch_single_pass<E: EthSpec>(
         }
     }
 
-    if conf.effective_balance_updates {
+    // Finish processing pending balance deposits if relevant.
-        state.set_total_active_balance(next_epoch, next_epoch_total_active_balance, spec);
+    //
-        *state.epoch_cache_mut() = next_epoch_cache.into_epoch_cache(
+    // This *could* be reordered after `process_pending_consolidations` which pushes only to the end
-            next_epoch_total_active_balance,
+    // of the `pending_balance_deposits` list. But we may as well preserve the write ordering used
-            next_epoch_activation_queue,
+    // 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>,
+    validator_info: &ValidatorInfo,
+    exit_cache: &mut ExitCache,
+    activation_queues: Option<(&BTreeSet<usize>, &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>,
     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<Validator>,
+    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<Validator>,
     exit_cache: &mut ExitCache,
@@ -568,6 +694,146 @@ fn process_single_slashing(
     Ok(())
 }
 
+impl PendingBalanceDepositsContext {
+    fn new<E: EthSpec>(state: &BeaconState<E>, spec: &ChainSpec) -> Result<Self, Error> {
+        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<u64>,
+    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<E: EthSpec>(
+    state: &mut BeaconState<E>,
+    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<Self, Error> {
         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>,
-    validator_info: &ValidatorInfo,
+    validator_current_epoch_participation: ParticipationFlags,
-    next_epoch_total_active_balance: &mut u64,
     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) {
+    let is_active_next_epoch = validator.is_active_at(state_ctxt.next_epoch);
-        next_epoch_total_active_balance.safe_add_assign(new_effective_balance)?;
-    }
 
     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.
+    // Caching: update next epoch effective balances and total active balance.
-    next_epoch_cache.push_effective_balance(new_effective_balance);
+    next_epoch_cache.update_effective_balance(
+        validator_index,
+        new_effective_balance,
+        is_active_next_epoch,
+    )?;
 
     Ok(())
 }

consensus/types/src/validator.rs

@@ -1,6 +1,6 @@
 use crate::{
-    test_utils::TestRandom, Address, BeaconState, ChainSpec, Epoch, EthSpec, ForkName, Hash256,
+    test_utils::TestRandom, Address, BeaconState, ChainSpec, Checkpoint, Epoch, EthSpec, ForkName,
-    PublicKeyBytes,
+    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<E: EthSpec>(
         &self,
         state: &BeaconState<E>,
         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 {