Redb optimize temp state cleanuop

beacon_node/beacon_chain/src/attestation_rewards.rs

@@ -11,7 +11,7 @@ use state_processing::per_epoch_processing::altair::{
 };
 use state_processing::per_epoch_processing::base::rewards_and_penalties::{
     get_attestation_component_delta, get_attestation_deltas_all, get_attestation_deltas_subset,
-    get_inactivity_penalty_delta, get_inclusion_delay_delta,
+    get_inactivity_penalty_delta, get_inclusion_delay_delta, ProposerRewardCalculation,
 };
 use state_processing::per_epoch_processing::base::validator_statuses::InclusionInfo;
 use state_processing::per_epoch_processing::base::{
@@ -81,13 +81,24 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             self.compute_ideal_rewards_base(&state, &validator_statuses.total_balances)?;
 
         let indices_to_attestation_delta = if validators.is_empty() {
-            get_attestation_deltas_all(&state, &validator_statuses, spec)?
-                .into_iter()
-                .enumerate()
-                .collect()
+            get_attestation_deltas_all(
+                &state,
+                &validator_statuses,
+                ProposerRewardCalculation::Exclude,
+                spec,
+            )?
+            .into_iter()
+            .enumerate()
+            .collect()
         } else {
             let validator_indices = Self::validators_ids_to_indices(&mut state, validators)?;
-            get_attestation_deltas_subset(&state, &validator_statuses, &validator_indices, spec)?
+            get_attestation_deltas_subset(
+                &state,
+                &validator_statuses,
+                ProposerRewardCalculation::Exclude,
+                &validator_indices,
+                spec,
+            )?
         };
 
         let mut total_rewards = vec![];

beacon_node/beacon_chain/src/beacon_block_reward.rs

@@ -1,28 +1,32 @@
-use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
+use crate::{BeaconChain, BeaconChainError, BeaconChainTypes, StateSkipConfig};
+use attesting_indices_base::get_attesting_indices;
 use eth2::lighthouse::StandardBlockReward;
-use operation_pool::RewardCache;
 use safe_arith::SafeArith;
 use slog::error;
+use state_processing::common::attesting_indices_base;
 use state_processing::{
-    common::{get_attestation_participation_flag_indices, get_attesting_indices_from_state},
+    common::{
+        base::{self, SqrtTotalActiveBalance},
+        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,
     },
 };
+use std::collections::HashSet;
 use store::{
     consts::altair::{PARTICIPATION_FLAG_WEIGHTS, PROPOSER_WEIGHT, WEIGHT_DENOMINATOR},
     RelativeEpoch,
 };
-use types::{AbstractExecPayload, BeaconBlockRef, BeaconState, BeaconStateError, Hash256};
+use types::{AbstractExecPayload, BeaconBlockRef, BeaconState, BeaconStateError, EthSpec};
 
 type BeaconBlockSubRewardValue = u64;
 
 impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn compute_beacon_block_reward<Payload: AbstractExecPayload<T::EthSpec>>(
         &self,
         block: BeaconBlockRef<'_, T::EthSpec, Payload>,
-        block_root: Hash256,
         state: &mut BeaconState<T::EthSpec>,
     ) -> Result<StandardBlockReward, BeaconChainError> {
         if block.slot() != state.slot() {
@@ -33,15 +37,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         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)
+        self.compute_beacon_block_reward_with_cache(block, state)
     }
 
     // This should only be called after a committee cache has been built
     // for both the previous and current epoch
     fn compute_beacon_block_reward_with_cache<Payload: AbstractExecPayload<T::EthSpec>>(
         &self,
         block: BeaconBlockRef<'_, T::EthSpec, Payload>,
-        block_root: Hash256,
         state: &BeaconState<T::EthSpec>,
     ) -> Result<StandardBlockReward, BeaconChainError> {
         let proposer_index = block.proposer_index();
@@ -72,7 +75,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             })?;
 
         let block_attestation_reward = if let BeaconState::Base(_) = state {
-            self.compute_beacon_block_attestation_reward_base(block, block_root, state)
+            self.compute_beacon_block_attestation_reward_base(block, state)
                 .map_err(|e| {
                     error!(
                         self.log,
@@ -169,19 +172,85 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     fn compute_beacon_block_attestation_reward_base<Payload: AbstractExecPayload<T::EthSpec>>(
         &self,
         block: BeaconBlockRef<'_, T::EthSpec, Payload>,
-        block_root: Hash256,
         state: &BeaconState<T::EthSpec>,
     ) -> Result<BeaconBlockSubRewardValue, BeaconChainError> {
-        // Call compute_block_reward in the base case
-        // Since base does not have sync aggregate, we only grab attesation portion of the returned
-        // value
-        let mut reward_cache = RewardCache::default();
-        let block_attestation_reward = self
-            .compute_block_reward(block, block_root, state, &mut reward_cache, true)?
-            .attestation_rewards
-            .total;
-
-        Ok(block_attestation_reward)
+        // In phase0, rewards for including attestations are awarded at epoch boundaries when the corresponding
+        // attestations are contained in state.previous_epoch_attestations. So, if an attestation within this block has
+        // target = previous_epoch, it is directly inserted into previous_epoch_attestations and we need the state at
+        // the end of this epoch, or the attestation has target = current_epoch and thus we need the state at the end
+        // of the next epoch.
+        // We fetch these lazily, as only one might be needed depending on the block's content.
+        let mut current_epoch_end = None;
+        let mut next_epoch_end = None;
+
+        let epoch = block.epoch();
+        let mut block_reward = 0;
+
+        let mut rewarded_attesters = HashSet::new();
+
+        for attestation in block.body().attestations() {
+            let processing_epoch_end = if attestation.data().target.epoch == epoch {
+                let next_epoch_end = match &mut next_epoch_end {
+                    Some(next_epoch_end) => next_epoch_end,
+                    None => {
+                        let state = self.state_at_slot(
+                            epoch.safe_add(1)?.end_slot(T::EthSpec::slots_per_epoch()),
+                            StateSkipConfig::WithoutStateRoots,
+                        )?;
+                        next_epoch_end.get_or_insert(state)
+                    }
+                };
+
+                // If the next epoch end is no longer phase0, no proposer rewards are awarded, as Altair epoch boundry
+                // processing kicks in. We check this here, as we know that current_epoch_end will always be phase0.
+                if !matches!(next_epoch_end, BeaconState::Base(_)) {
+                    continue;
+                }
+
+                next_epoch_end
+            } else if attestation.data().target.epoch == epoch.safe_sub(1)? {
+                match &mut current_epoch_end {
+                    Some(current_epoch_end) => current_epoch_end,
+                    None => {
+                        let state = self.state_at_slot(
+                            epoch.end_slot(T::EthSpec::slots_per_epoch()),
+                            StateSkipConfig::WithoutStateRoots,
+                        )?;
+                        current_epoch_end.get_or_insert(state)
+                    }
+                }
+            } else {
+                return Err(BeaconChainError::BlockRewardAttestationError);
+            };
+
+            let inclusion_delay = state.slot().safe_sub(attestation.data().slot)?.as_u64();
+            let sqrt_total_active_balance =
+                SqrtTotalActiveBalance::new(processing_epoch_end.get_total_active_balance()?);
+            for attester in get_attesting_indices_from_state(state, attestation)? {
+                let validator = processing_epoch_end.get_validator(attester as usize)?;
+                if !validator.slashed
+                    && !rewarded_attesters.contains(&attester)
+                    && !has_earlier_attestation(
+                        state,
+                        processing_epoch_end,
+                        inclusion_delay,
+                        attester,
+                    )?
+                {
+                    let base_reward = base::get_base_reward(
+                        validator.effective_balance,
+                        sqrt_total_active_balance,
+                        &self.spec,
+                    )?;
+                    let proposer_reward =
+                        base_reward.safe_div(self.spec.proposer_reward_quotient)?;
+                    block_reward.safe_add_assign(proposer_reward)?;
+                    rewarded_attesters.insert(attester);
+                }
+            }
+        }
+
+        Ok(block_reward)
     }
 
     fn compute_beacon_block_attestation_reward_altair_deneb<
@@ -244,3 +313,25 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         Ok(total_proposer_reward)
     }
 }
+
+fn has_earlier_attestation<E: EthSpec>(
+    state: &BeaconState<E>,
+    processing_epoch_end: &BeaconState<E>,
+    inclusion_delay: u64,
+    attester: u64,
+) -> Result<bool, BeaconChainError> {
+    if inclusion_delay > 1 {
+        for epoch_att in processing_epoch_end.previous_epoch_attestations()? {
+            if epoch_att.inclusion_delay < inclusion_delay {
+                let committee =
+                    state.get_beacon_committee(epoch_att.data.slot, epoch_att.data.index)?;
+                let earlier_attesters =
+                    get_attesting_indices::<E>(committee.committee, &epoch_att.aggregation_bits)?;
+                if earlier_attesters.contains(&attester) {
+                    return Ok(true);
+                }
+            }
+        }
+    }
+    Ok(false)
+}

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -5640,7 +5640,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let mut ctxt = ConsensusContext::new(block.slot());
 
         let consensus_block_value = self
-            .compute_beacon_block_reward(block.message(), Hash256::zero(), &mut state)
+            .compute_beacon_block_reward(block.message(), &mut state)
             .map(|reward| reward.total)
             .unwrap_or(0);
 

beacon_node/beacon_chain/src/eth1_chain.rs

@@ -475,10 +475,10 @@ impl<E: EthSpec> Eth1ChainBackend<E> for CachingEth1Backend<E> {
             voting_period_start_slot,
         );
 
-        let blocks = self.core.blocks().read();
-
-        let votes_to_consider =
-            get_votes_to_consider(blocks.iter(), voting_period_start_seconds, spec);
+        let votes_to_consider = {
+            let blocks = self.core.blocks().read();
+            get_votes_to_consider(blocks.iter(), voting_period_start_seconds, spec)
+        };
 
         trace!(
             self.log,

beacon_node/beacon_chain/src/migrate.rs

@@ -679,11 +679,9 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
                     StoreOp::DeleteSyncCommitteeBranch(block_root),
                 ]
             })
-            .chain(
-                abandoned_states
-                    .into_iter()
-                    .map(|(slot, state_hash)| StoreOp::DeleteState(state_hash.into(), Some(slot))),
-            )
+            .chain(abandoned_states.into_iter().map(|(slot, state_hash)| {
+                StoreOp::DeleteStateAndSummary(state_hash.into(), Some(slot))
+            }))
             .collect();
 
         // Persist the head in case the process is killed or crashes here. This prevents