Implement block_rewards API (per-validator reward) (#3907)

## Issue Addressed

[#3661](#3661)

## Proposed Changes

`/eth/v1/beacon/rewards/blocks/{block_id}`

```
{
  "execution_optimistic": false,
  "finalized": false,
  "data": {
    "proposer_index": "123",
    "total": "123",
    "attestations": "123",
    "sync_aggregate": "123",
    "proposer_slashings": "123",
    "attester_slashings": "123"
  }
}
```

The issue contains the implementation of three per-validator reward APIs:
* `sync_committee_rewards`
* [`attestation_rewards`](#3822)
* `block_rewards`

This PR only implements the `block_rewards`.

The endpoints can be viewed in the Ethereum Beacon nodes API browser: [https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Rewards](https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Rewards)

## Additional Info

The implementation of [consensus client reward APIs](https://github.com/eth-protocol-fellows/cohort-three/blob/master/projects/project-ideas.md#consensus-client-reward-apis) is part of the [EPF](https://github.com/eth-protocol-fellows/cohort-three).

Co-authored-by: kevinbogner <kevbogner@gmail.com>
Co-authored-by: navie <naviechan@gmail.com>

beacon_node/beacon_chain/src/beacon_block_reward.rs

@@ -0,0 +1,237 @@
+use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
+use eth2::lighthouse::StandardBlockReward;
+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,
+    },
+    per_block_processing::{
+        altair::sync_committee::compute_sync_aggregate_rewards, get_slashable_indices,
+    },
+};
+use store::{
+    consts::altair::{PARTICIPATION_FLAG_WEIGHTS, PROPOSER_WEIGHT, WEIGHT_DENOMINATOR},
+    RelativeEpoch,
+};
+use types::{BeaconBlockRef, BeaconState, BeaconStateError, ExecPayload, Hash256};
+
+type BeaconBlockSubRewardValue = u64;
+
+impl<T: BeaconChainTypes> BeaconChain<T> {
+    pub fn compute_beacon_block_reward<Payload: ExecPayload<T::EthSpec>>(
+        &self,
+        block: BeaconBlockRef<'_, T::EthSpec, Payload>,
+        block_root: Hash256,
+        state: &mut BeaconState<T::EthSpec>,
+    ) -> Result<StandardBlockReward, BeaconChainError> {
+        if block.slot() != state.slot() {
+            return Err(BeaconChainError::BlockRewardSlotError);
+        }
+
+        state.build_committee_cache(RelativeEpoch::Previous, &self.spec)?;
+        state.build_committee_cache(RelativeEpoch::Current, &self.spec)?;
+
+        let proposer_index = block.proposer_index();
+
+        let sync_aggregate_reward =
+            self.compute_beacon_block_sync_aggregate_reward(block, state)?;
+
+        let proposer_slashing_reward = self
+            .compute_beacon_block_proposer_slashing_reward(block, state)
+            .map_err(|e| {
+                error!(
+                self.log,
+                "Error calculating proposer slashing reward";
+                "error" => ?e
+                );
+                BeaconChainError::BlockRewardError
+            })?;
+
+        let attester_slashing_reward = self
+            .compute_beacon_block_attester_slashing_reward(block, state)
+            .map_err(|e| {
+                error!(
+                self.log,
+                "Error calculating attester slashing reward";
+                "error" => ?e
+                );
+                BeaconChainError::BlockRewardError
+            })?;
+
+        let block_attestation_reward = if let BeaconState::Base(_) = state {
+            self.compute_beacon_block_attestation_reward_base(block, block_root, state)
+                .map_err(|e| {
+                    error!(
+                    self.log,
+                    "Error calculating base block attestation reward";
+                    "error" => ?e
+                    );
+                    BeaconChainError::BlockRewardAttestationError
+                })?
+        } else {
+            self.compute_beacon_block_attestation_reward_altair(block, state)
+                .map_err(|e| {
+                    error!(
+                    self.log,
+                    "Error calculating altair block attestation reward";
+                    "error" => ?e
+                    );
+                    BeaconChainError::BlockRewardAttestationError
+                })?
+        };
+
+        let total_reward = sync_aggregate_reward
+            .safe_add(proposer_slashing_reward)?
+            .safe_add(attester_slashing_reward)?
+            .safe_add(block_attestation_reward)?;
+
+        Ok(StandardBlockReward {
+            proposer_index,
+            total: total_reward,
+            attestations: block_attestation_reward,
+            sync_aggregate: sync_aggregate_reward,
+            proposer_slashings: proposer_slashing_reward,
+            attester_slashings: attester_slashing_reward,
+        })
+    }
+
+    fn compute_beacon_block_sync_aggregate_reward<Payload: ExecPayload<T::EthSpec>>(
+        &self,
+        block: BeaconBlockRef<'_, T::EthSpec, Payload>,
+        state: &BeaconState<T::EthSpec>,
+    ) -> Result<BeaconBlockSubRewardValue, BeaconChainError> {
+        if let Ok(sync_aggregate) = block.body().sync_aggregate() {
+            let (_, proposer_reward_per_bit) = compute_sync_aggregate_rewards(state, &self.spec)
+                .map_err(|_| BeaconChainError::BlockRewardSyncError)?;
+            Ok(sync_aggregate.sync_committee_bits.num_set_bits() as u64 * proposer_reward_per_bit)
+        } else {
+            Ok(0)
+        }
+    }
+
+    fn compute_beacon_block_proposer_slashing_reward<Payload: ExecPayload<T::EthSpec>>(
+        &self,
+        block: BeaconBlockRef<'_, T::EthSpec, Payload>,
+        state: &BeaconState<T::EthSpec>,
+    ) -> Result<BeaconBlockSubRewardValue, BeaconChainError> {
+        let mut proposer_slashing_reward = 0;
+
+        let proposer_slashings = block.body().proposer_slashings();
+
+        for proposer_slashing in proposer_slashings {
+            proposer_slashing_reward.safe_add_assign(
+                state
+                    .get_validator(proposer_slashing.proposer_index() as usize)?
+                    .effective_balance
+                    .safe_div(self.spec.whistleblower_reward_quotient)?,
+            )?;
+        }
+
+        Ok(proposer_slashing_reward)
+    }
+
+    fn compute_beacon_block_attester_slashing_reward<Payload: ExecPayload<T::EthSpec>>(
+        &self,
+        block: BeaconBlockRef<'_, T::EthSpec, Payload>,
+        state: &BeaconState<T::EthSpec>,
+    ) -> Result<BeaconBlockSubRewardValue, BeaconChainError> {
+        let mut attester_slashing_reward = 0;
+
+        let attester_slashings = block.body().attester_slashings();
+
+        for attester_slashing in attester_slashings {
+            for attester_index in get_slashable_indices(state, attester_slashing)? {
+                attester_slashing_reward.safe_add_assign(
+                    state
+                        .get_validator(attester_index as usize)?
+                        .effective_balance
+                        .safe_div(self.spec.whistleblower_reward_quotient)?,
+                )?;
+            }
+        }
+
+        Ok(attester_slashing_reward)
+    }
+
+    fn compute_beacon_block_attestation_reward_base<Payload: ExecPayload<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)
+    }
+
+    fn compute_beacon_block_attestation_reward_altair<Payload: ExecPayload<T::EthSpec>>(
+        &self,
+        block: BeaconBlockRef<'_, T::EthSpec, Payload>,
+        state: &mut 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
+            .safe_sub(PROPOSER_WEIGHT)?
+            .safe_mul(WEIGHT_DENOMINATOR)?
+            .safe_div(PROPOSER_WEIGHT)?;
+
+        for attestation in block.body().attestations() {
+            let data = &attestation.data;
+            let inclusion_delay = state.slot().safe_sub(data.slot)?.as_u64();
+            let participation_flag_indices = get_attestation_participation_flag_indices(
+                state,
+                data,
+                inclusion_delay,
+                &self.spec,
+            )?;
+
+            let attesting_indices = get_attesting_indices_from_state(state, attestation)?;
+
+            let mut proposer_reward_numerator = 0;
+            for index in attesting_indices {
+                let index = index as usize;
+                for (flag_index, &weight) in PARTICIPATION_FLAG_WEIGHTS.iter().enumerate() {
+                    let epoch_participation =
+                        state.get_epoch_participation_mut(data.target.epoch)?;
+                    let validator_participation = epoch_participation
+                        .get_mut(index)
+                        .ok_or(BeaconStateError::ParticipationOutOfBounds(index))?;
+
+                    if participation_flag_indices.contains(&flag_index)
+                        && !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)?,
+                        )?;
+                    }
+                }
+            }
+            total_proposer_reward.safe_add_assign(
+                proposer_reward_numerator.safe_div(proposer_reward_denominator)?,
+            )?;
+        }
+
+        Ok(total_proposer_reward)
+    }
+}

beacon_node/beacon_chain/src/errors.rs

@@ -154,6 +154,7 @@ pub enum BeaconChainError {
     ExecutionForkChoiceUpdateInvalid {
         status: PayloadStatus,
     },
+    BlockRewardError,
     BlockRewardSlotError,
     BlockRewardAttestationError,
     BlockRewardSyncError,

beacon_node/beacon_chain/src/lib.rs

@@ -2,6 +2,7 @@
 pub mod attestation_rewards;
 pub mod attestation_verification;
 mod attester_cache;
+pub mod beacon_block_reward;
 mod beacon_chain;
 mod beacon_fork_choice_store;
 pub mod beacon_proposer_cache;

beacon_node/http_api/src/lib.rs

@@ -15,6 +15,7 @@ mod database;
 mod metrics;
 mod proposer_duties;
 mod publish_blocks;
+mod standard_block_rewards;
 mod state_id;
 mod sync_committee_rewards;
 mod sync_committees;
@@ -1700,6 +1701,27 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    let beacon_rewards_path = eth_v1
+        .and(warp::path("beacon"))
+        .and(warp::path("rewards"))
+        .and(chain_filter.clone());
+
+    // GET beacon/rewards/blocks/{block_id}
+    let get_beacon_rewards_blocks = beacon_rewards_path
+        .clone()
+        .and(warp::path("blocks"))
+        .and(block_id_or_err)
+        .and(warp::path::end())
+        .and_then(|chain: Arc<BeaconChain<T>>, block_id: BlockId| {
+            blocking_json_task(move || {
+                let (rewards, execution_optimistic) =
+                    standard_block_rewards::compute_beacon_block_rewards(chain, block_id)?;
+                Ok(rewards)
+                    .map(api_types::GenericResponse::from)
+                    .map(|resp| resp.add_execution_optimistic(execution_optimistic))
+            })
+        });
+
     /*
      * beacon/rewards
      */
@@ -3433,6 +3455,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(get_beacon_pool_proposer_slashings.boxed())
                 .or(get_beacon_pool_voluntary_exits.boxed())
                 .or(get_beacon_deposit_snapshot.boxed())
+                .or(get_beacon_rewards_blocks.boxed())
                 .or(get_config_fork_schedule.boxed())
                 .or(get_config_spec.boxed())
                 .or(get_config_deposit_contract.boxed())

beacon_node/http_api/src/standard_block_rewards.rs

@@ -0,0 +1,27 @@
+use crate::sync_committee_rewards::get_state_before_applying_block;
+use crate::BlockId;
+use crate::ExecutionOptimistic;
+use beacon_chain::{BeaconChain, BeaconChainTypes};
+use eth2::lighthouse::StandardBlockReward;
+use std::sync::Arc;
+use warp_utils::reject::beacon_chain_error;
+//// The difference between block_rewards and beacon_block_rewards is the later returns block
+//// reward format that satisfies beacon-api specs
+pub fn compute_beacon_block_rewards<T: BeaconChainTypes>(
+    chain: Arc<BeaconChain<T>>,
+    block_id: BlockId,
+) -> Result<(StandardBlockReward, ExecutionOptimistic), warp::Rejection> {
+    let (block, execution_optimistic) = block_id.blinded_block(&chain)?;
+
+    let block_ref = block.message();
+
+    let block_root = block.canonical_root();
+
+    let mut state = get_state_before_applying_block(chain.clone(), &block)?;
+
+    let rewards = chain
+        .compute_beacon_block_reward(block_ref, block_root, &mut state)
+        .map_err(beacon_chain_error)?;
+
+    Ok((rewards, execution_optimistic))
+}

beacon_node/http_api/src/sync_committee_rewards.rs

@@ -47,7 +47,7 @@ pub fn compute_sync_committee_rewards<T: BeaconChainTypes>(
     Ok((data, execution_optimistic))
 }
 
-fn get_state_before_applying_block<T: BeaconChainTypes>(
+pub fn get_state_before_applying_block<T: BeaconChainTypes>(
     chain: Arc<BeaconChain<T>>,
     block: &SignedBlindedBeaconBlock<T::EthSpec>,
 ) -> Result<BeaconState<T::EthSpec>, warp::reject::Rejection> {

beacon_node/operation_pool/src/lib.rs

@@ -8,7 +8,7 @@ mod persistence;
 mod reward_cache;
 mod sync_aggregate_id;
 
-pub use attestation::AttMaxCover;
+pub use attestation::{earliest_attestation_validators, AttMaxCover};
 pub use attestation_storage::{AttestationRef, SplitAttestation};
 pub use max_cover::MaxCover;
 pub use persistence::{

common/eth2/src/lib.rs

@@ -1044,6 +1044,22 @@ impl BeaconNodeHttpClient {
         Ok(())
     }
 
+    /// `GET beacon/rewards/blocks`
+    pub async fn get_beacon_rewards_blocks(&self, epoch: Epoch) -> Result<(), Error> {
+        let mut path = self.eth_path(V1)?;
+
+        path.path_segments_mut()
+            .map_err(|()| Error::InvalidUrl(self.server.clone()))?
+            .push("beacon")
+            .push("rewards")
+            .push("blocks");
+
+        path.query_pairs_mut()
+            .append_pair("epoch", &epoch.to_string());
+
+        self.get(path).await
+    }
+
     /// `POST beacon/rewards/attestations`
     pub async fn post_beacon_rewards_attestations(
         &self,

common/eth2/src/lighthouse.rs

@@ -4,6 +4,7 @@ mod attestation_performance;
 pub mod attestation_rewards;
 mod block_packing_efficiency;
 mod block_rewards;
+mod standard_block_rewards;
 mod sync_committee_rewards;
 
 use crate::{
@@ -30,6 +31,7 @@ pub use block_packing_efficiency::{
 };
 pub use block_rewards::{AttestationRewards, BlockReward, BlockRewardMeta, BlockRewardsQuery};
 pub use lighthouse_network::{types::SyncState, PeerInfo};
+pub use standard_block_rewards::StandardBlockReward;
 pub use sync_committee_rewards::SyncCommitteeReward;
 
 // Define "legacy" implementations of `Option<T>` which use four bytes for encoding the union