Implement sync_committee_rewards API (per-validator reward) (#3903)

[#3661](https://github.com/sigp/lighthouse/issues/3661)

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

```
{
  "execution_optimistic": false,
  "finalized": false,
  "data": [
    {
      "validator_index": "0",
      "reward": "2000"
    }
  ]
}
```

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

This PR only implements the `sync_committe_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)

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: navie <naviechan@gmail.com>
Co-authored-by: kevinbogner <kevbogner@gmail.com>

beacon_node/http_api/src/lib.rs

@@ -16,6 +16,7 @@ mod metrics;
 mod proposer_duties;
 mod publish_blocks;
 mod state_id;
+mod sync_committee_rewards;
 mod sync_committees;
 mod ui;
 mod validator_inclusion;
@@ -1794,6 +1795,41 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    /*
+     * beacon/rewards
+     */
+
+    let beacon_rewards_path = eth_v1
+        .and(warp::path("beacon"))
+        .and(warp::path("rewards"))
+        .and(chain_filter.clone());
+
+    // POST beacon/rewards/sync_committee/{block_id}
+    let post_beacon_rewards_sync_committee = beacon_rewards_path
+        .clone()
+        .and(warp::path("sync_committee"))
+        .and(block_id_or_err)
+        .and(warp::path::end())
+        .and(warp::body::json())
+        .and(log_filter.clone())
+        .and_then(
+            |chain: Arc<BeaconChain<T>>,
+             block_id: BlockId,
+             validators: Vec<ValidatorId>,
+             log: Logger| {
+                blocking_json_task(move || {
+                    let (rewards, execution_optimistic) =
+                        sync_committee_rewards::compute_sync_committee_rewards(
+                            chain, block_id, validators, log,
+                        )?;
+
+                    Ok(rewards)
+                        .map(api_types::GenericResponse::from)
+                        .map(|resp| resp.add_execution_optimistic(execution_optimistic))
+                })
+            },
+        );
+
     /*
      * config
      */
@@ -3528,6 +3564,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_sync_committee.boxed())
                 .or(post_beacon_pool_bls_to_execution_changes.boxed())
                 .or(post_validator_duties_attester.boxed())
                 .or(post_validator_duties_sync.boxed())

beacon_node/http_api/src/sync_committee_rewards.rs

@@ -0,0 +1,77 @@
+use crate::{BlockId, ExecutionOptimistic};
+use beacon_chain::{BeaconChain, BeaconChainError, BeaconChainTypes};
+use eth2::lighthouse::SyncCommitteeReward;
+use eth2::types::ValidatorId;
+use slog::{debug, Logger};
+use state_processing::BlockReplayer;
+use std::sync::Arc;
+use types::{BeaconState, SignedBlindedBeaconBlock};
+use warp_utils::reject::{beacon_chain_error, custom_not_found};
+
+pub fn compute_sync_committee_rewards<T: BeaconChainTypes>(
+    chain: Arc<BeaconChain<T>>,
+    block_id: BlockId,
+    validators: Vec<ValidatorId>,
+    log: Logger,
+) -> Result<(Option<Vec<SyncCommitteeReward>>, ExecutionOptimistic), warp::Rejection> {
+    let (block, execution_optimistic) = block_id.blinded_block(&chain)?;
+
+    let mut state = get_state_before_applying_block(chain.clone(), &block)?;
+
+    let reward_payload = chain
+        .compute_sync_committee_rewards(block.message(), &mut state)
+        .map_err(beacon_chain_error)?;
+
+    let data = if reward_payload.is_empty() {
+        debug!(log, "compute_sync_committee_rewards returned empty");
+        None
+    } else if validators.is_empty() {
+        Some(reward_payload)
+    } else {
+        Some(
+            reward_payload
+                .into_iter()
+                .filter(|reward| {
+                    validators.iter().any(|validator| match validator {
+                        ValidatorId::Index(i) => reward.validator_index == *i,
+                        ValidatorId::PublicKey(pubkey) => match state.get_validator_index(pubkey) {
+                            Ok(Some(i)) => reward.validator_index == i as u64,
+                            _ => false,
+                        },
+                    })
+                })
+                .collect::<Vec<SyncCommitteeReward>>(),
+        )
+    };
+
+    Ok((data, execution_optimistic))
+}
+
+fn get_state_before_applying_block<T: BeaconChainTypes>(
+    chain: Arc<BeaconChain<T>>,
+    block: &SignedBlindedBeaconBlock<T::EthSpec>,
+) -> Result<BeaconState<T::EthSpec>, warp::reject::Rejection> {
+    let parent_block: SignedBlindedBeaconBlock<T::EthSpec> = chain
+        .get_blinded_block(&block.parent_root())
+        .and_then(|maybe_block| {
+            maybe_block.ok_or_else(|| BeaconChainError::MissingBeaconBlock(block.parent_root()))
+        })
+        .map_err(|e| custom_not_found(format!("Parent block is not available! {:?}", e)))?;
+
+    let parent_state = chain
+        .get_state(&parent_block.state_root(), Some(parent_block.slot()))
+        .and_then(|maybe_state| {
+            maybe_state
+                .ok_or_else(|| BeaconChainError::MissingBeaconState(parent_block.state_root()))
+        })
+        .map_err(|e| custom_not_found(format!("Parent state is not available! {:?}", e)))?;
+
+    let replayer = BlockReplayer::new(parent_state, &chain.spec)
+        .no_signature_verification()
+        .state_root_iter([Ok((parent_block.state_root(), parent_block.slot()))].into_iter())
+        .minimal_block_root_verification()
+        .apply_blocks(vec![], Some(block.slot()))
+        .map_err(beacon_chain_error)?;
+
+    Ok(replayer.into_state())
+}