Implement attestation_rewards API (per-validator reward) (#3822)

## Issue Addressed

#3661 

## Proposed Changes
`/eth/v1/beacon/rewards/attestations/{epoch}`

```json
{
  "execution_optimistic": false,
  "finalized": false,
  "data": [
    {
      "ideal_rewards": [
        {
          "effective_balance": "1000000000",
          "head": "2500",
          "target": "5000",
          "source": "5000"
        }
      ],
      "total_rewards": [
        {
          "validator_index": "0",
          "head": "2000",
          "target": "2000",
          "source": "4000",
          "inclusion_delay": "2000"
        }
      ]
    }
  ]
}
```

The issue contains the implementation of three per-validator reward APIs:
- [`sync_committee_rewards`](https://github.com/sigp/lighthouse/pull/3790)
- `attestation_rewards`
- `block_rewards`.

This PR *only* implements the `attestation_rewards`.

The endpoints can be viewed in the Ethereum Beacon nodes API browser: 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).

---
- [x] `get_state`
- [x] Calculate *ideal rewards* with some logic from  `get_flag_index_deltas`
- [x] Calculate *actual rewards*  with some logic from `get_flag_index_deltas`
- [x] Code cleanup
- [x] Testing

beacon_node/beacon_chain/src/attestation_rewards.rs

@@ -0,0 +1,196 @@
+use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
+use eth2::lighthouse::attestation_rewards::{IdealAttestationRewards, TotalAttestationRewards};
+use eth2::lighthouse::StandardAttestationRewards;
+use participation_cache::ParticipationCache;
+use safe_arith::SafeArith;
+use slog::{debug, Logger};
+use state_processing::{
+    common::altair::BaseRewardPerIncrement,
+    per_epoch_processing::altair::{participation_cache, rewards_and_penalties::get_flag_weight},
+};
+use std::collections::HashMap;
+use store::consts::altair::{
+    PARTICIPATION_FLAG_WEIGHTS, TIMELY_HEAD_FLAG_INDEX, TIMELY_SOURCE_FLAG_INDEX,
+    TIMELY_TARGET_FLAG_INDEX,
+};
+use types::consts::altair::WEIGHT_DENOMINATOR;
+
+use types::{Epoch, EthSpec};
+
+use eth2::types::ValidatorId;
+
+impl<T: BeaconChainTypes> BeaconChain<T> {
+    pub fn compute_attestation_rewards(
+        &self,
+        epoch: Epoch,
+        validators: Vec<ValidatorId>,
+        log: Logger,
+    ) -> Result<StandardAttestationRewards, BeaconChainError> {
+        debug!(log, "computing attestation rewards"; "epoch" => epoch, "validator_count" => validators.len());
+
+        // Get state
+        let spec = &self.spec;
+
+        let state_slot = (epoch + 1).end_slot(T::EthSpec::slots_per_epoch());
+
+        let state_root = self
+            .state_root_at_slot(state_slot)?
+            .ok_or(BeaconChainError::NoStateForSlot(state_slot))?;
+
+        let mut state = self
+            .get_state(&state_root, Some(state_slot))?
+            .ok_or(BeaconChainError::MissingBeaconState(state_root))?;
+
+        // Calculate ideal_rewards
+        let participation_cache = ParticipationCache::new(&state, spec)?;
+
+        let previous_epoch = state.previous_epoch();
+
+        let mut ideal_rewards_hashmap = HashMap::new();
+
+        for flag_index in 0..PARTICIPATION_FLAG_WEIGHTS.len() {
+            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_increments =
+                unslashed_participating_balance.safe_div(spec.effective_balance_increment)?;
+
+            let total_active_balance = participation_cache.current_epoch_total_active_balance();
+
+            let active_increments =
+                total_active_balance.safe_div(spec.effective_balance_increment)?;
+
+            let base_reward_per_increment =
+                BaseRewardPerIncrement::new(total_active_balance, spec)?;
+
+            for effective_balance_eth in 0..=32 {
+                let base_reward =
+                    effective_balance_eth.safe_mul(base_reward_per_increment.as_u64())?;
+
+                let penalty = -(base_reward.safe_mul(weight)?.safe_div(WEIGHT_DENOMINATOR)? as i64);
+
+                let reward_numerator = base_reward
+                    .safe_mul(weight)?
+                    .safe_mul(unslashed_participating_increments)?;
+
+                let ideal_reward = reward_numerator
+                    .safe_div(active_increments)?
+                    .safe_div(WEIGHT_DENOMINATOR)?;
+                if !state.is_in_inactivity_leak(previous_epoch, spec) {
+                    ideal_rewards_hashmap
+                        .insert((flag_index, effective_balance_eth), (ideal_reward, penalty));
+                } else {
+                    ideal_rewards_hashmap.insert((flag_index, effective_balance_eth), (0, penalty));
+                }
+            }
+        }
+
+        // Calculate total_rewards
+        let mut total_rewards: Vec<TotalAttestationRewards> = Vec::new();
+
+        let validators = if validators.is_empty() {
+            participation_cache.eligible_validator_indices().to_vec()
+        } else {
+            validators
+                .into_iter()
+                .map(|validator| match validator {
+                    ValidatorId::Index(i) => Ok(i as usize),
+                    ValidatorId::PublicKey(pubkey) => state
+                        .get_validator_index(&pubkey)?
+                        .ok_or(BeaconChainError::ValidatorPubkeyUnknown(pubkey)),
+                })
+                .collect::<Result<Vec<_>, _>>()?
+        };
+
+        for validator_index in &validators {
+            let eligible = state.is_eligible_validator(previous_epoch, *validator_index)?;
+            let mut head_reward = 0u64;
+            let mut target_reward = 0i64;
+            let mut source_reward = 0i64;
+
+            if eligible {
+                let effective_balance = state.get_effective_balance(*validator_index)?;
+
+                let effective_balance_eth =
+                    effective_balance.safe_div(spec.effective_balance_increment)?;
+
+                for flag_index in 0..PARTICIPATION_FLAG_WEIGHTS.len() {
+                    let (ideal_reward, penalty) = ideal_rewards_hashmap
+                        .get(&(flag_index, effective_balance_eth))
+                        .ok_or(BeaconChainError::AttestationRewardsError)?;
+                    let voted_correctly = participation_cache
+                        .get_unslashed_participating_indices(flag_index, previous_epoch)
+                        .map_err(|_| BeaconChainError::AttestationRewardsError)?
+                        .contains(*validator_index)
+                        .map_err(|_| BeaconChainError::AttestationRewardsError)?;
+                    if voted_correctly {
+                        if flag_index == TIMELY_HEAD_FLAG_INDEX {
+                            head_reward += ideal_reward;
+                        } else if flag_index == TIMELY_TARGET_FLAG_INDEX {
+                            target_reward += *ideal_reward as i64;
+                        } else if flag_index == TIMELY_SOURCE_FLAG_INDEX {
+                            source_reward += *ideal_reward as i64;
+                        }
+                    } else if flag_index == TIMELY_HEAD_FLAG_INDEX {
+                        head_reward = 0;
+                    } else if flag_index == TIMELY_TARGET_FLAG_INDEX {
+                        target_reward = *penalty;
+                    } else if flag_index == TIMELY_SOURCE_FLAG_INDEX {
+                        source_reward = *penalty;
+                    }
+                }
+            }
+            total_rewards.push(TotalAttestationRewards {
+                validator_index: *validator_index as u64,
+                head: head_reward,
+                target: target_reward,
+                source: source_reward,
+            });
+        }
+
+        // Convert hashmap to vector
+        let mut ideal_rewards: Vec<IdealAttestationRewards> = ideal_rewards_hashmap
+            .iter()
+            .map(
+                |((flag_index, effective_balance_eth), (ideal_reward, _penalty))| {
+                    (flag_index, effective_balance_eth, ideal_reward)
+                },
+            )
+            .fold(
+                HashMap::new(),
+                |mut acc, (flag_index, effective_balance_eth, ideal_reward)| {
+                    let entry = acc.entry(*effective_balance_eth as u32).or_insert(
+                        IdealAttestationRewards {
+                            effective_balance: *effective_balance_eth,
+                            head: 0,
+                            target: 0,
+                            source: 0,
+                        },
+                    );
+                    match *flag_index {
+                        TIMELY_SOURCE_FLAG_INDEX => entry.source += ideal_reward,
+                        TIMELY_TARGET_FLAG_INDEX => entry.target += ideal_reward,
+                        TIMELY_HEAD_FLAG_INDEX => entry.head += ideal_reward,
+                        _ => {}
+                    }
+                    acc
+                },
+            )
+            .into_values()
+            .collect::<Vec<IdealAttestationRewards>>();
+        ideal_rewards.sort_by(|a, b| a.effective_balance.cmp(&b.effective_balance));
+
+        Ok(StandardAttestationRewards {
+            ideal_rewards,
+            total_rewards,
+        })
+    }
+}

beacon_node/beacon_chain/src/errors.rs

@@ -50,7 +50,6 @@ pub enum BeaconChainError {
     },
     SlotClockDidNotStart,
     NoStateForSlot(Slot),
-    UnableToFindTargetRoot(Slot),
     BeaconStateError(BeaconStateError),
     DBInconsistent(String),
     DBError(store::Error),
@@ -159,6 +158,7 @@ pub enum BeaconChainError {
     BlockRewardAttestationError,
     BlockRewardSyncError,
     SyncCommitteeRewardsSyncError,
+    AttestationRewardsError,
     HeadMissingFromForkChoice(Hash256),
     FinalizedBlockMissingFromForkChoice(Hash256),
     HeadBlockMissingFromForkChoice(Hash256),

beacon_node/beacon_chain/src/lib.rs

@@ -1,4 +1,5 @@
 #![recursion_limit = "128"] // For lazy-static
+pub mod attestation_rewards;
 pub mod attestation_verification;
 mod attester_cache;
 mod beacon_chain;

beacon_node/http_api/src/lib.rs

@@ -1709,6 +1709,58 @@ pub fn serve<T: BeaconChainTypes>(
         .and(warp::path("rewards"))
         .and(chain_filter.clone());
 
+    // POST beacon/rewards/attestations/{epoch}
+    let post_beacon_rewards_attestations = beacon_rewards_path
+        .clone()
+        .and(warp::path("attestations"))
+        .and(warp::path::param::<Epoch>())
+        .and(warp::path::end())
+        .and(warp::body::json())
+        .and(log_filter.clone())
+        .and_then(
+            |chain: Arc<BeaconChain<T>>,
+             epoch: Epoch,
+             validators: Vec<ValidatorId>,
+             log: Logger| {
+                blocking_json_task(move || {
+                    let attestation_rewards = chain
+                        .compute_attestation_rewards(epoch, validators, log)
+                        .map_err(|e| match e {
+                            BeaconChainError::MissingBeaconState(root) => {
+                                warp_utils::reject::custom_not_found(format!(
+                                    "missing state {root:?}",
+                                ))
+                            }
+                            BeaconChainError::NoStateForSlot(slot) => {
+                                warp_utils::reject::custom_not_found(format!(
+                                    "missing state at slot {slot}"
+                                ))
+                            }
+                            BeaconChainError::BeaconStateError(
+                                BeaconStateError::UnknownValidator(validator_index),
+                            ) => warp_utils::reject::custom_bad_request(format!(
+                                "validator is unknown: {validator_index}"
+                            )),
+                            BeaconChainError::ValidatorPubkeyUnknown(pubkey) => {
+                                warp_utils::reject::custom_bad_request(format!(
+                                    "validator pubkey is unknown: {pubkey:?}"
+                                ))
+                            }
+                            e => warp_utils::reject::custom_server_error(format!(
+                                "unexpected error: {:?}",
+                                e
+                            )),
+                        })?;
+                    let execution_optimistic =
+                        chain.is_optimistic_or_invalid_head().unwrap_or_default();
+
+                    Ok(attestation_rewards)
+                        .map(api_types::GenericResponse::from)
+                        .map(|resp| resp.add_execution_optimistic(execution_optimistic))
+                })
+            },
+        );
+
     // POST beacon/rewards/sync_committee/{block_id}
     let post_beacon_rewards_sync_committee = beacon_rewards_path
         .clone()
@@ -3432,6 +3484,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(post_beacon_pool_proposer_slashings.boxed())
                 .or(post_beacon_pool_voluntary_exits.boxed())
                 .or(post_beacon_pool_sync_committees.boxed())
+                .or(post_beacon_rewards_attestations.boxed())
                 .or(post_beacon_rewards_sync_committee.boxed())
                 .or(post_validator_duties_attester.boxed())
                 .or(post_validator_duties_sync.boxed())

common/eth2/src/lib.rs

@@ -1044,6 +1044,24 @@ impl BeaconNodeHttpClient {
         Ok(())
     }
 
+    /// `POST beacon/rewards/attestations`
+    pub async fn post_beacon_rewards_attestations(
+        &self,
+        attestations: &[ValidatorId],
+    ) -> 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("attestations");
+
+        self.post(path, &attestations).await?;
+
+        Ok(())
+    }
+
     /// `POST validator/contribution_and_proofs`
     pub async fn post_validator_contribution_and_proofs<T: EthSpec>(
         &self,

common/eth2/src/lighthouse.rs

@@ -1,6 +1,7 @@
 //! This module contains endpoints that are non-standard and only available on Lighthouse servers.
 
 mod attestation_performance;
+pub mod attestation_rewards;
 mod block_packing_efficiency;
 mod block_rewards;
 mod sync_committee_rewards;
@@ -23,6 +24,7 @@ use store::{AnchorInfo, Split, StoreConfig};
 pub use attestation_performance::{
     AttestationPerformance, AttestationPerformanceQuery, AttestationPerformanceStatistics,
 };
+pub use attestation_rewards::StandardAttestationRewards;
 pub use block_packing_efficiency::{
     BlockPackingEfficiency, BlockPackingEfficiencyQuery, ProposerInfo, UniqueAttestation,
 };

common/eth2/src/lighthouse/attestation_rewards.rs

@@ -0,0 +1,42 @@
+use serde::{Deserialize, Serialize};
+
+// Details about the rewards paid for attestations
+// All rewards in GWei
+
+#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
+pub struct IdealAttestationRewards {
+    // Validator's effective balance in gwei
+    #[serde(with = "eth2_serde_utils::quoted_u64")]
+    pub effective_balance: u64,
+    // Ideal attester's reward for head vote in gwei
+    #[serde(with = "eth2_serde_utils::quoted_u64")]
+    pub head: u64,
+    // Ideal attester's reward for target vote in gwei
+    #[serde(with = "eth2_serde_utils::quoted_u64")]
+    pub target: u64,
+    // Ideal attester's reward for source vote in gwei
+    #[serde(with = "eth2_serde_utils::quoted_u64")]
+    pub source: u64,
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
+pub struct TotalAttestationRewards {
+    // one entry for every validator based on their attestations in the epoch
+    #[serde(with = "eth2_serde_utils::quoted_u64")]
+    pub validator_index: u64,
+    // attester's reward for head vote in gwei
+    #[serde(with = "eth2_serde_utils::quoted_u64")]
+    pub head: u64,
+    // attester's reward for target vote in gwei
+    pub target: i64,
+    // attester's reward for source vote in gwei
+    pub source: i64,
+    // TBD attester's inclusion_delay reward in gwei (phase0 only)
+    // pub inclusion_delay: u64,
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
+pub struct StandardAttestationRewards {
+    pub ideal_rewards: Vec<IdealAttestationRewards>,
+    pub total_rewards: Vec<TotalAttestationRewards>,
+}

common/eth2/src/types.rs

@@ -270,11 +270,20 @@ pub struct FinalityCheckpointsData {
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(try_from = "&str")]
 pub enum ValidatorId {
     PublicKey(PublicKeyBytes),
     Index(u64),
 }
 
+impl TryFrom<&str> for ValidatorId {
+    type Error = String;
+
+    fn try_from(s: &str) -> Result<Self, Self::Error> {
+        Self::from_str(s)
+    }
+}
+
 impl FromStr for ValidatorId {
     type Err = String;