Implement API for block rewards (#2628)

## Proposed Changes

Add an API endpoint for retrieving detailed information about block rewards.

For information on usage see [the docs](https://github.com/sigp/lighthouse/blob/block-rewards-api/book/src/api-lighthouse.md#lighthouseblock_rewards), and the source.

beacon_node/beacon_chain/src/block_reward.rs

@@ -0,0 +1,97 @@
+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, Hash256, RelativeEpoch};
+
+impl<T: BeaconChainTypes> BeaconChain<T> {
+    pub fn compute_block_reward(
+        &self,
+        block: BeaconBlockRef<'_, T::EthSpec>,
+        block_root: Hash256,
+        state: &BeaconState<T::EthSpec>,
+    ) -> 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_balance(active_indices, &self.spec)?;
+        let mut per_attestation_rewards = block
+            .body()
+            .attestations()
+            .iter()
+            .map(|att| {
+                AttMaxCover::new(att, state, total_active_balance, &self.spec)
+                    .ok_or(BeaconChainError::BlockRewardAttestationError)
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+
+        // Update the attestation rewards for each previous attestation included.
+        // This is O(n^2) in the number of attestations n.
+        for i in 0..per_attestation_rewards.len() {
+            let (updated, to_update) = per_attestation_rewards.split_at_mut(i + 1);
+            let latest_att = &updated[i];
+
+            for att in to_update {
+                att.update_covering_set(latest_att.object(), latest_att.covering_set());
+            }
+        }
+
+        let mut prev_epoch_total = 0;
+        let mut curr_epoch_total = 0;
+
+        for cover in &per_attestation_rewards {
+            for &reward in cover.fresh_validators_rewards.values() {
+                if cover.att.data.slot.epoch(T::EthSpec::slots_per_epoch()) == state.current_epoch()
+                {
+                    curr_epoch_total += reward;
+                } else {
+                    prev_epoch_total += reward;
+                }
+            }
+        }
+
+        let attestation_total = prev_epoch_total + curr_epoch_total;
+
+        // Drop the covers.
+        let per_attestation_rewards = per_attestation_rewards
+            .into_iter()
+            .map(|cover| cover.fresh_validators_rewards)
+            .collect();
+
+        let attestation_rewards = AttestationRewards {
+            total: attestation_total,
+            prev_epoch_total,
+            curr_epoch_total,
+            per_attestation_rewards,
+        };
+
+        // Sync committee rewards.
+        let sync_committee_rewards = 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)?;
+            sync_aggregate.sync_committee_bits.num_set_bits() as u64 * proposer_reward_per_bit
+        } else {
+            0
+        };
+
+        // Total, metadata
+        let total = attestation_total + sync_committee_rewards;
+
+        let meta = BlockRewardMeta {
+            slot: block.slot(),
+            parent_slot: state.latest_block_header().slot,
+            proposer_index: block.proposer_index(),
+            graffiti: block.body().graffiti().as_utf8_lossy(),
+        };
+
+        Ok(BlockReward {
+            total,
+            block_root,
+            meta,
+            attestation_rewards,
+            sync_committee_rewards,
+        })
+    }
+}

beacon_node/beacon_chain/src/block_verification.rs

@@ -53,6 +53,7 @@ use crate::{
     },
     metrics, BeaconChain, BeaconChainError, BeaconChainTypes,
 };
+use eth2::types::EventKind;
 use fork_choice::{ForkChoice, ForkChoiceStore, PayloadVerificationStatus};
 use parking_lot::RwLockReadGuard;
 use proto_array::Block as ProtoBlock;
@@ -1165,6 +1166,18 @@ impl<'a, T: BeaconChainTypes> FullyVerifiedBlock<'a, T> {
 
         metrics::stop_timer(committee_timer);
 
+        /*
+         * If we have block reward listeners, compute the block reward and push it to the
+         * event handler.
+         */
+        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)?;
+                event_handler.register(EventKind::BlockReward(block_reward));
+            }
+        }
+
         /*
          * Perform `per_block_processing` on the block and state, returning early if the block is
          * invalid.

beacon_node/beacon_chain/src/errors.rs

@@ -137,6 +137,9 @@ pub enum BeaconChainError {
     AltairForkDisabled,
     ExecutionLayerMissing,
     ExecutionForkChoiceUpdateFailed(execution_layer::Error),
+    BlockRewardSlotError,
+    BlockRewardAttestationError,
+    BlockRewardSyncError,
     HeadMissingFromForkChoice(Hash256),
     FinalizedBlockMissingFromForkChoice(Hash256),
     InvalidFinalizedPayloadShutdownError(TrySendError<ShutdownReason>),

beacon_node/beacon_chain/src/events.rs

@@ -15,6 +15,7 @@ pub struct ServerSentEventHandler<T: EthSpec> {
     chain_reorg_tx: Sender<EventKind<T>>,
     contribution_tx: Sender<EventKind<T>>,
     late_head: Sender<EventKind<T>>,
+    block_reward_tx: Sender<EventKind<T>>,
     log: Logger,
 }
 
@@ -32,6 +33,7 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         let (chain_reorg_tx, _) = broadcast::channel(capacity);
         let (contribution_tx, _) = broadcast::channel(capacity);
         let (late_head, _) = broadcast::channel(capacity);
+        let (block_reward_tx, _) = broadcast::channel(capacity);
 
         Self {
             attestation_tx,
@@ -42,6 +44,7 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
             chain_reorg_tx,
             contribution_tx,
             late_head,
+            block_reward_tx,
             log,
         }
     }
@@ -67,6 +70,8 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
                 .map(|count| trace!(self.log, "Registering server-sent contribution and proof event"; "receiver_count" => count)),
             EventKind::LateHead(late_head) => self.late_head.send(EventKind::LateHead(late_head))
                 .map(|count| trace!(self.log, "Registering server-sent late head event"; "receiver_count" => count)),
+            EventKind::BlockReward(block_reward) => self.block_reward_tx.send(EventKind::BlockReward(block_reward))
+                .map(|count| trace!(self.log, "Registering server-sent contribution and proof event"; "receiver_count" => count)),
         };
         if let Err(SendError(event)) = result {
             trace!(self.log, "No receivers registered to listen for event"; "event" => ?event);
@@ -105,6 +110,10 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         self.late_head.subscribe()
     }
 
+    pub fn subscribe_block_reward(&self) -> Receiver<EventKind<T>> {
+        self.block_reward_tx.subscribe()
+    }
+
     pub fn has_attestation_subscribers(&self) -> bool {
         self.attestation_tx.receiver_count() > 0
     }
@@ -136,4 +145,8 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
     pub fn has_late_head_subscribers(&self) -> bool {
         self.late_head.receiver_count() > 0
     }
+
+    pub fn has_block_reward_subscribers(&self) -> bool {
+        self.block_reward_tx.receiver_count() > 0
+    }
 }

beacon_node/beacon_chain/src/lib.rs

@@ -5,6 +5,7 @@ mod beacon_chain;
 mod beacon_fork_choice_store;
 mod beacon_proposer_cache;
 mod beacon_snapshot;
+pub mod block_reward;
 mod block_times_cache;
 mod block_verification;
 pub mod builder;

beacon_node/http_api/src/block_rewards.rs

@@ -0,0 +1,80 @@
+use beacon_chain::{BeaconChain, BeaconChainError, BeaconChainTypes, WhenSlotSkipped};
+use eth2::lighthouse::{BlockReward, BlockRewardsQuery};
+use slog::{warn, Logger};
+use state_processing::BlockReplayer;
+use std::sync::Arc;
+use warp_utils::reject::{beacon_chain_error, beacon_state_error, custom_bad_request};
+
+pub fn get_block_rewards<T: BeaconChainTypes>(
+    query: BlockRewardsQuery,
+    chain: Arc<BeaconChain<T>>,
+    log: Logger,
+) -> Result<Vec<BlockReward>, warp::Rejection> {
+    let start_slot = query.start_slot;
+    let end_slot = query.end_slot;
+    let prior_slot = start_slot - 1;
+
+    if start_slot > end_slot || start_slot == 0 {
+        return Err(custom_bad_request(format!(
+            "invalid start and end: {}, {}",
+            start_slot, end_slot
+        )));
+    }
+
+    let end_block_root = chain
+        .block_root_at_slot(end_slot, WhenSlotSkipped::Prev)
+        .map_err(beacon_chain_error)?
+        .ok_or_else(|| custom_bad_request(format!("block at end slot {} unknown", end_slot)))?;
+
+    let blocks = chain
+        .store
+        .load_blocks_to_replay(start_slot, end_slot, end_block_root)
+        .map_err(|e| beacon_chain_error(e.into()))?;
+
+    let state_root = chain
+        .state_root_at_slot(prior_slot)
+        .map_err(beacon_chain_error)?
+        .ok_or_else(|| custom_bad_request(format!("prior state at slot {} unknown", prior_slot)))?;
+
+    let mut state = chain
+        .get_state(&state_root, Some(prior_slot))
+        .and_then(|maybe_state| maybe_state.ok_or(BeaconChainError::MissingBeaconState(state_root)))
+        .map_err(beacon_chain_error)?;
+
+    state
+        .build_all_caches(&chain.spec)
+        .map_err(beacon_state_error)?;
+
+    let mut block_rewards = Vec::with_capacity(blocks.len());
+
+    let block_replayer = BlockReplayer::new(state, &chain.spec)
+        .pre_block_hook(Box::new(|state, block| {
+            // Compute block reward.
+            let block_reward =
+                chain.compute_block_reward(block.message(), block.canonical_root(), state)?;
+            block_rewards.push(block_reward);
+            Ok(())
+        }))
+        .state_root_iter(
+            chain
+                .forwards_iter_state_roots_until(prior_slot, end_slot)
+                .map_err(beacon_chain_error)?,
+        )
+        .no_signature_verification()
+        .minimal_block_root_verification()
+        .apply_blocks(blocks, None)
+        .map_err(beacon_chain_error)?;
+
+    if block_replayer.state_root_miss() {
+        warn!(
+            log,
+            "Block reward state root miss";
+            "start_slot" => start_slot,
+            "end_slot" => end_slot,
+        );
+    }
+
+    drop(block_replayer);
+
+    Ok(block_rewards)
+}

beacon_node/http_api/src/lib.rs

@@ -7,6 +7,7 @@
 
 mod attester_duties;
 mod block_id;
+mod block_rewards;
 mod database;
 mod metrics;
 mod proposer_duties;
@@ -2540,6 +2541,16 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    let get_lighthouse_block_rewards = warp::path("lighthouse")
+        .and(warp::path("block_rewards"))
+        .and(warp::query::<eth2::lighthouse::BlockRewardsQuery>())
+        .and(warp::path::end())
+        .and(chain_filter.clone())
+        .and(log_filter.clone())
+        .and_then(|query, chain, log| {
+            blocking_json_task(move || block_rewards::get_block_rewards(query, chain, log))
+        });
+
     let get_events = eth1_v1
         .and(warp::path("events"))
         .and(warp::path::end())
@@ -2576,6 +2587,9 @@ pub fn serve<T: BeaconChainTypes>(
                                 api_types::EventTopic::LateHead => {
                                     event_handler.subscribe_late_head()
                                 }
+                                api_types::EventTopic::BlockReward => {
+                                    event_handler.subscribe_block_reward()
+                                }
                             };
 
                             receivers.push(BroadcastStream::new(receiver).map(|msg| {
@@ -2661,6 +2675,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(get_lighthouse_beacon_states_ssz.boxed())
                 .or(get_lighthouse_staking.boxed())
                 .or(get_lighthouse_database_info.boxed())
+                .or(get_lighthouse_block_rewards.boxed())
                 .or(get_events.boxed()),
         )
         .or(warp::post().and(

beacon_node/operation_pool/src/lib.rs

@@ -6,15 +6,16 @@ mod metrics;
 mod persistence;
 mod sync_aggregate_id;
 
+pub use attestation::AttMaxCover;
+pub use max_cover::MaxCover;
 pub use persistence::{
     PersistedOperationPool, PersistedOperationPoolAltair, PersistedOperationPoolBase,
 };
 
 use crate::sync_aggregate_id::SyncAggregateId;
-use attestation::AttMaxCover;
 use attestation_id::AttestationId;
 use attester_slashing::AttesterSlashingMaxCover;
-use max_cover::{maximum_cover, MaxCover};
+use max_cover::maximum_cover;
 use parking_lot::RwLock;
 use state_processing::per_block_processing::errors::AttestationValidationError;
 use state_processing::per_block_processing::{