Extend block reward APIs (#3290)

## Proposed Changes

Add a new HTTP endpoint `POST /lighthouse/analysis/block_rewards` which takes a vec of `BeaconBlock`s as input and outputs the `BlockReward`s for them.

Augment the `BlockReward` struct with the attestation data for attestations in the block, which simplifies access to this information from blockprint. Using attestation data I've been able to make blockprint up to 95% accurate across Prysm/Lighthouse/Teku/Nimbus. I hope to go even higher using a bunch of synthetic blocks produced for Prysm/Nimbus/Lodestar, which are underrepresented in the current training data.

Cargo.lock

@@ -2521,6 +2521,7 @@ dependencies = [
  "lighthouse_network",
  "lighthouse_version",
  "logging",
+ "lru",
  "network",
  "parking_lot 0.12.1",
  "safe_arith",

beacon_node/beacon_chain/src/block_reward.rs

@@ -2,7 +2,7 @@ use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
 use eth2::lighthouse::{AttestationRewards, BlockReward, BlockRewardMeta};
 use operation_pool::{AttMaxCover, MaxCover};
 use state_processing::per_block_processing::altair::sync_committee::compute_sync_aggregate_rewards;
-use types::{BeaconBlockRef, BeaconState, EthSpec, ExecPayload, Hash256, RelativeEpoch};
+use types::{BeaconBlockRef, BeaconState, EthSpec, ExecPayload, Hash256};
 
 impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn compute_block_reward<Payload: ExecPayload<T::EthSpec>>(
@@ -10,13 +10,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         block: BeaconBlockRef<'_, T::EthSpec, Payload>,
         block_root: Hash256,
         state: &BeaconState<T::EthSpec>,
+        include_attestations: bool,
     ) -> Result<BlockReward, BeaconChainError> {
         if block.slot() != state.slot() {
             return Err(BeaconChainError::BlockRewardSlotError);
         }
 
-        let active_indices = state.get_cached_active_validator_indices(RelativeEpoch::Current)?;
+        let total_active_balance = state.get_total_active_balance()?;
-        let total_active_balance = state.get_total_balance(active_indices, &self.spec)?;
         let mut per_attestation_rewards = block
             .body()
             .attestations()
@@ -60,11 +60,24 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .map(|cover| cover.fresh_validators_rewards)
             .collect();
 
+        // Add the attestation data if desired.
+        let attestations = if include_attestations {
+            block
+                .body()
+                .attestations()
+                .iter()
+                .map(|a| a.data.clone())
+                .collect()
+        } else {
+            vec![]
+        };
+
         let attestation_rewards = AttestationRewards {
             total: attestation_total,
             prev_epoch_total,
             curr_epoch_total,
             per_attestation_rewards,
+            attestations,
         };
 
         // Sync committee rewards.

beacon_node/beacon_chain/src/block_verification.rs

@@ -1235,7 +1235,7 @@ impl<'a, T: BeaconChainTypes> FullyVerifiedBlock<'a, T> {
         if let Some(ref event_handler) = chain.event_handler {
             if event_handler.has_block_reward_subscribers() {
                 let block_reward =
-                    chain.compute_block_reward(block.message(), block_root, &state)?;
+                    chain.compute_block_reward(block.message(), block_root, &state, true)?;
                 event_handler.register(EventKind::BlockReward(block_reward));
             }
         }

beacon_node/http_api/Cargo.toml

@@ -31,7 +31,7 @@ execution_layer = {path = "../execution_layer"}
 parking_lot = "0.12.0"
 safe_arith = {path = "../../consensus/safe_arith"}
 task_executor = { path = "../../common/task_executor" }
-
+lru = "0.7.7"
 
 [dev-dependencies]
 store = { path = "../store" }

beacon_node/http_api/src/block_rewards.rs

@@ -1,10 +1,17 @@
 use beacon_chain::{BeaconChain, BeaconChainError, BeaconChainTypes, WhenSlotSkipped};
 use eth2::lighthouse::{BlockReward, BlockRewardsQuery};
-use slog::{warn, Logger};
+use lru::LruCache;
+use slog::{debug, warn, Logger};
 use state_processing::BlockReplayer;
 use std::sync::Arc;
-use warp_utils::reject::{beacon_chain_error, beacon_state_error, custom_bad_request};
+use types::BeaconBlock;
+use warp_utils::reject::{
+    beacon_chain_error, beacon_state_error, custom_bad_request, custom_server_error,
+};
 
+const STATE_CACHE_SIZE: usize = 2;
+
+/// Fetch block rewards for blocks from the canonical chain.
 pub fn get_block_rewards<T: BeaconChainTypes>(
     query: BlockRewardsQuery,
     chain: Arc<BeaconChain<T>>,
@@ -50,8 +57,12 @@ pub fn get_block_rewards<T: BeaconChainTypes>(
     let block_replayer = BlockReplayer::new(state, &chain.spec)
         .pre_block_hook(Box::new(|state, block| {
             // Compute block reward.
-            let block_reward =
+            let block_reward = chain.compute_block_reward(
-                chain.compute_block_reward(block.message(), block.canonical_root(), state)?;
+                block.message(),
+                block.canonical_root(),
+                state,
+                query.include_attestations,
+            )?;
             block_rewards.push(block_reward);
             Ok(())
         }))
@@ -78,3 +89,84 @@ pub fn get_block_rewards<T: BeaconChainTypes>(
 
     Ok(block_rewards)
 }
+
+/// Compute block rewards for blocks passed in as input.
+pub fn compute_block_rewards<T: BeaconChainTypes>(
+    blocks: Vec<BeaconBlock<T::EthSpec>>,
+    chain: Arc<BeaconChain<T>>,
+    log: Logger,
+) -> Result<Vec<BlockReward>, warp::Rejection> {
+    let mut block_rewards = Vec::with_capacity(blocks.len());
+    let mut state_cache = LruCache::new(STATE_CACHE_SIZE);
+
+    for block in blocks {
+        let parent_root = block.parent_root();
+
+        // Check LRU cache for a constructed state from a previous iteration.
+        let state = if let Some(state) = state_cache.get(&(parent_root, block.slot())) {
+            debug!(
+                log,
+                "Re-using cached state for block rewards";
+                "parent_root" => ?parent_root,
+                "slot" => block.slot(),
+            );
+            state
+        } else {
+            debug!(
+                log,
+                "Fetching state for block rewards";
+                "parent_root" => ?parent_root,
+                "slot" => block.slot()
+            );
+            let parent_block = chain
+                .get_blinded_block(&parent_root)
+                .map_err(beacon_chain_error)?
+                .ok_or_else(|| {
+                    custom_bad_request(format!(
+                        "parent block not known or not canonical: {:?}",
+                        parent_root
+                    ))
+                })?;
+
+            let parent_state = chain
+                .get_state(&parent_block.state_root(), Some(parent_block.slot()))
+                .map_err(beacon_chain_error)?
+                .ok_or_else(|| {
+                    custom_bad_request(format!(
+                        "no state known for parent block: {:?}",
+                        parent_root
+                    ))
+                })?;
+
+            let block_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)?;
+
+            if block_replayer.state_root_miss() {
+                warn!(
+                    log,
+                    "Block reward state root miss";
+                    "parent_slot" => parent_block.slot(),
+                    "slot" => block.slot(),
+                );
+            }
+
+            state_cache
+                .get_or_insert((parent_root, block.slot()), || block_replayer.into_state())
+                .ok_or_else(|| {
+                    custom_server_error("LRU cache insert should always succeed".into())
+                })?
+        };
+
+        // Compute block reward.
+        let block_reward = chain
+            .compute_block_reward(block.to_ref(), block.canonical_root(), state, true)
+            .map_err(beacon_chain_error)?;
+        block_rewards.push(block_reward);
+    }
+
+    Ok(block_rewards)
+}

beacon_node/http_api/src/lib.rs

@@ -2823,6 +2823,18 @@ pub fn serve<T: BeaconChainTypes>(
             blocking_json_task(move || block_rewards::get_block_rewards(query, chain, log))
         });
 
+    // POST lighthouse/analysis/block_rewards
+    let post_lighthouse_block_rewards = warp::path("lighthouse")
+        .and(warp::path("analysis"))
+        .and(warp::path("block_rewards"))
+        .and(warp::body::json())
+        .and(warp::path::end())
+        .and(chain_filter.clone())
+        .and(log_filter.clone())
+        .and_then(|blocks, chain, log| {
+            blocking_json_task(move || block_rewards::compute_block_rewards(blocks, chain, log))
+        });
+
     // GET lighthouse/analysis/attestation_performance/{index}
     let get_lighthouse_attestation_performance = warp::path("lighthouse")
         .and(warp::path("analysis"))
@@ -2998,7 +3010,8 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(post_validator_prepare_beacon_proposer.boxed())
                 .or(post_lighthouse_liveness.boxed())
                 .or(post_lighthouse_database_reconstruct.boxed())
-                .or(post_lighthouse_database_historical_blocks.boxed()),
+                .or(post_lighthouse_database_historical_blocks.boxed())
+                .or(post_lighthouse_block_rewards.boxed()),
         ))
         .recover(warp_utils::reject::handle_rejection)
         .with(slog_logging(log.clone()))

common/eth2/src/lighthouse/block_rewards.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use types::{Hash256, Slot};
+use types::{AttestationData, Hash256, Slot};
 
 /// Details about the rewards paid to a block proposer for proposing a block.
 ///
@@ -42,6 +42,9 @@ pub struct AttestationRewards {
     ///
     /// Each element of the vec is a map from validator index to reward.
     pub per_attestation_rewards: Vec<HashMap<u64, u64>>,
+    /// The attestations themselves (optional).
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub attestations: Vec<AttestationData>,
 }
 
 /// Query parameters for the `/lighthouse/block_rewards` endpoint.
@@ -51,4 +54,7 @@ pub struct BlockRewardsQuery {
     pub start_slot: Slot,
     /// Upper slot limit for block rewards returned (inclusive).
     pub end_slot: Slot,
+    /// Include the full attestations themselves?
+    #[serde(default)]
+    pub include_attestations: bool,
 }