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

## Issue Addressed

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

## Proposed Changes

`/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)

## 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: navie <naviechan@gmail.com>
Co-authored-by: kevinbogner <kevbogner@gmail.com>
This commit is contained in:
naviechan 2023-01-24 02:06:42 +00:00
parent 3e67fa3038
commit 2802bc9a9c
11 changed files with 382 additions and 0 deletions

View File

@ -160,6 +160,7 @@ pub enum BeaconChainError {
BlockRewardSlotError,
BlockRewardAttestationError,
BlockRewardSyncError,
SyncCommitteeRewardsSyncError,
HeadMissingFromForkChoice(Hash256),
FinalizedBlockMissingFromForkChoice(Hash256),
HeadBlockMissingFromForkChoice(Hash256),

View File

@ -40,6 +40,7 @@ pub mod schema_change;
mod shuffling_cache;
mod snapshot_cache;
pub mod state_advance_timer;
pub mod sync_committee_rewards;
pub mod sync_committee_verification;
pub mod test_utils;
mod timeout_rw_lock;

View File

@ -0,0 +1,87 @@
use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
use eth2::lighthouse::SyncCommitteeReward;
use safe_arith::SafeArith;
use slog::error;
use state_processing::per_block_processing::altair::sync_committee::compute_sync_aggregate_rewards;
use std::collections::HashMap;
use store::RelativeEpoch;
use types::{BeaconBlockRef, BeaconState, ExecPayload};
impl<T: BeaconChainTypes> BeaconChain<T> {
pub fn compute_sync_committee_rewards<Payload: ExecPayload<T::EthSpec>>(
&self,
block: BeaconBlockRef<'_, T::EthSpec, Payload>,
state: &mut BeaconState<T::EthSpec>,
) -> Result<Vec<SyncCommitteeReward>, BeaconChainError> {
if block.slot() != state.slot() {
return Err(BeaconChainError::BlockRewardSlotError);
}
let spec = &self.spec;
state.build_committee_cache(RelativeEpoch::Current, spec)?;
let sync_aggregate = block.body().sync_aggregate()?;
let sync_committee = state.current_sync_committee()?.clone();
let sync_committee_indices = state.get_sync_committee_indices(&sync_committee)?;
let (participant_reward_value, proposer_reward_per_bit) =
compute_sync_aggregate_rewards(state, spec).map_err(|e| {
error!(
self.log, "Error calculating sync aggregate rewards";
"error" => ?e
);
BeaconChainError::SyncCommitteeRewardsSyncError
})?;
let mut balances = HashMap::<usize, u64>::new();
let mut total_proposer_rewards = 0;
let proposer_index = state.get_beacon_proposer_index(block.slot(), spec)?;
// Apply rewards to participant balances. Keep track of proposer rewards
for (validator_index, participant_bit) in sync_committee_indices
.iter()
.zip(sync_aggregate.sync_committee_bits.iter())
{
let participant_balance = balances
.entry(*validator_index)
.or_insert_with(|| state.balances()[*validator_index]);
if participant_bit {
participant_balance.safe_add_assign(participant_reward_value)?;
balances
.entry(proposer_index)
.or_insert_with(|| state.balances()[proposer_index])
.safe_add_assign(proposer_reward_per_bit)?;
total_proposer_rewards.safe_add_assign(proposer_reward_per_bit)?;
} else {
*participant_balance = participant_balance.saturating_sub(participant_reward_value);
}
}
Ok(balances
.iter()
.filter_map(|(i, new_balance)| {
let reward = if *i != proposer_index {
*new_balance as i64 - state.balances()[*i] as i64
} else if sync_committee_indices.contains(i) {
*new_balance as i64
- state.balances()[*i] as i64
- total_proposer_rewards as i64
} else {
return None;
};
Some(SyncCommitteeReward {
validator_index: *i as u64,
reward,
})
})
.collect())
}
}

View File

@ -2,6 +2,7 @@ pub use crate::persisted_beacon_chain::PersistedBeaconChain;
pub use crate::{
beacon_chain::{BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, FORK_CHOICE_DB_KEY, OP_POOL_DB_KEY},
migrate::MigratorConfig,
sync_committee_verification::Error as SyncCommitteeError,
validator_monitor::DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD,
BeaconChainError, NotifyExecutionLayer, ProduceBlockVerification,
};
@ -1980,6 +1981,30 @@ where
(honest_head, faulty_head)
}
pub fn process_sync_contributions(
&self,
sync_contributions: HarnessSyncContributions<E>,
) -> Result<(), SyncCommitteeError> {
let mut verified_contributions = Vec::with_capacity(sync_contributions.len());
for (_, contribution_and_proof) in sync_contributions {
let signed_contribution_and_proof = contribution_and_proof.unwrap();
let verified_contribution = self
.chain
.verify_sync_contribution_for_gossip(signed_contribution_and_proof)?;
verified_contributions.push(verified_contribution);
}
for verified_contribution in verified_contributions {
self.chain
.add_contribution_to_block_inclusion_pool(verified_contribution)?;
}
Ok(())
}
}
// Junk `Debug` impl to satistfy certain trait bounds during testing.

View File

@ -4,6 +4,7 @@ mod block_verification;
mod merge;
mod op_verification;
mod payload_invalidation;
mod rewards;
mod store_tests;
mod sync_committee_verification;
mod tests;

View File

@ -0,0 +1,121 @@
#![cfg(test)]
use std::collections::HashMap;
use beacon_chain::test_utils::{
generate_deterministic_keypairs, BeaconChainHarness, EphemeralHarnessType,
};
use beacon_chain::{
test_utils::{AttestationStrategy, BlockStrategy, RelativeSyncCommittee},
types::{Epoch, EthSpec, Keypair, MinimalEthSpec},
};
use lazy_static::lazy_static;
pub const VALIDATOR_COUNT: usize = 64;
lazy_static! {
static ref KEYPAIRS: Vec<Keypair> = generate_deterministic_keypairs(VALIDATOR_COUNT);
}
fn get_harness<E: EthSpec>() -> BeaconChainHarness<EphemeralHarnessType<E>> {
let mut spec = E::default_spec();
spec.altair_fork_epoch = Some(Epoch::new(0)); // We use altair for all tests
let harness = BeaconChainHarness::builder(E::default())
.spec(spec)
.keypairs(KEYPAIRS.to_vec())
.fresh_ephemeral_store()
.build();
harness.advance_slot();
harness
}
#[tokio::test]
async fn test_sync_committee_rewards() {
let num_block_produced = MinimalEthSpec::slots_per_epoch();
let harness = get_harness::<MinimalEthSpec>();
let latest_block_root = harness
.extend_chain(
num_block_produced as usize,
BlockStrategy::OnCanonicalHead,
AttestationStrategy::AllValidators,
)
.await;
// Create and add sync committee message to op_pool
let sync_contributions = harness.make_sync_contributions(
&harness.get_current_state(),
latest_block_root,
harness.get_current_slot(),
RelativeSyncCommittee::Current,
);
harness
.process_sync_contributions(sync_contributions)
.unwrap();
// Add block
let chain = &harness.chain;
let (head_state, head_state_root) = harness.get_current_state_and_root();
let target_slot = harness.get_current_slot() + 1;
let (block_root, mut state) = harness
.add_attested_block_at_slot(target_slot, head_state, head_state_root, &[])
.await
.unwrap();
let block = harness.get_block(block_root).unwrap();
let parent_block = chain
.get_blinded_block(&block.parent_root())
.unwrap()
.unwrap();
let parent_state = chain
.get_state(&parent_block.state_root(), Some(parent_block.slot()))
.unwrap()
.unwrap();
let reward_payload = chain
.compute_sync_committee_rewards(block.message(), &mut state)
.unwrap();
let rewards = reward_payload
.iter()
.map(|reward| (reward.validator_index, reward.reward))
.collect::<HashMap<_, _>>();
let proposer_index = state
.get_beacon_proposer_index(target_slot, &MinimalEthSpec::default_spec())
.unwrap();
let mut mismatches = vec![];
for validator in state.validators() {
let validator_index = state
.clone()
.get_validator_index(&validator.pubkey)
.unwrap()
.unwrap();
let pre_state_balance = parent_state.balances()[validator_index];
let post_state_balance = state.balances()[validator_index];
let sync_committee_reward = rewards.get(&(validator_index as u64)).unwrap_or(&0);
if validator_index == proposer_index {
continue; // Ignore proposer
}
if pre_state_balance as i64 + *sync_committee_reward != post_state_balance as i64 {
mismatches.push(validator_index.to_string());
}
}
assert_eq!(
mismatches.len(),
0,
"Expect 0 mismatches, but these validators have mismatches on balance: {} ",
mismatches.join(",")
);
}

View File

@ -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;
@ -1699,6 +1700,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
*/
@ -3396,6 +3432,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_validator_duties_attester.boxed())
.or(post_validator_duties_sync.boxed())
.or(post_validator_aggregate_and_proofs.boxed())

View File

@ -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())
}

View File

@ -1026,6 +1026,24 @@ impl BeaconNodeHttpClient {
.transpose()
}
/// `POST beacon/rewards/sync_committee`
pub async fn post_beacon_rewards_sync_committee(
&self,
rewards: &[Option<Vec<lighthouse::SyncCommitteeReward>>],
) -> 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("sync_committee");
self.post(path, &rewards).await?;
Ok(())
}
/// `POST validator/contribution_and_proofs`
pub async fn post_validator_contribution_and_proofs<T: EthSpec>(
&self,

View File

@ -3,6 +3,7 @@
mod attestation_performance;
mod block_packing_efficiency;
mod block_rewards;
mod sync_committee_rewards;
use crate::{
ok_or_error,
@ -27,6 +28,7 @@ pub use block_packing_efficiency::{
};
pub use block_rewards::{AttestationRewards, BlockReward, BlockRewardMeta, BlockRewardsQuery};
pub use lighthouse_network::{types::SyncState, PeerInfo};
pub use sync_committee_rewards::SyncCommitteeReward;
// Define "legacy" implementations of `Option<T>` which use four bytes for encoding the union
// selector.

View File

@ -0,0 +1,12 @@
use serde::{Deserialize, Serialize};
// Details about the rewards paid to sync committee members for attesting headers
// All rewards in GWei
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
pub struct SyncCommitteeReward {
#[serde(with = "eth2_serde_utils::quoted_u64")]
pub validator_index: u64,
// sync committee reward in gwei for the validator
pub reward: i64,
}