Merge branch 'unstable' into deneb-free-blobs

.github/workflows/test-suite.yml

@@ -17,6 +17,10 @@ env:
   PINNED_NIGHTLY: nightly-2023-04-16
   # Prevent Github API rate limiting.
   LIGHTHOUSE_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+  # Enable self-hosted runners for the sigp repo only.
+  SELF_HOSTED_RUNNERS: ${{ github.repository == 'sigp/lighthouse' }}
+  # Self-hosted runners need to reference a different host for `./watch` tests.
+  WATCH_HOST: ${{ github.repository == 'sigp/lighthouse' && 'host.docker.internal' || 'localhost' }}
 jobs:
   target-branch-check:
     name: target-branch-check
@@ -48,11 +52,13 @@ jobs:
       run: make cargo-fmt
   release-tests-ubuntu:
     name: release-tests-ubuntu
-    runs-on: ubuntu-latest
+    # Use self-hosted runners only on the sigp repo.
+    runs-on: ${{ github.repository == 'sigp/lighthouse' && fromJson('["self-hosted", "linux", "large"]') || 'ubuntu-latest'  }}
     needs: cargo-fmt
     steps:
     - uses: actions/checkout@v3
     - name: Get latest version of stable Rust
+      if: env.SELF_HOSTED_RUNNERS == false
       run: rustup update stable
     - name: Install Protoc
       uses: arduino/setup-protoc@e52d9eb8f7b63115df1ac544a1376fdbf5a39612
@@ -64,11 +70,12 @@ jobs:
       run: make test-release
   release-tests-windows:
     name: release-tests-windows
-    runs-on: windows-2019
+    runs-on: ${{ github.repository == 'sigp/lighthouse' && fromJson('["self-hosted", "windows"]') || 'windows-2019'  }}
     needs: cargo-fmt
     steps:
     - uses: actions/checkout@v3
     - name: Get latest version of stable Rust
+      if: env.SELF_HOSTED_RUNNERS == false
       run: rustup update stable
     - name: Use Node.js
       uses: actions/setup-node@v2
@@ -83,6 +90,7 @@ jobs:
     - name: Install make
       run: choco install -y make
     - uses: KyleMayes/install-llvm-action@v1
+      if: env.SELF_HOSTED_RUNNERS == false
       with:
         version: "15.0"
         directory: ${{ runner.temp }}/llvm
@@ -92,11 +100,13 @@ jobs:
       run: make test-release
   beacon-chain-tests:
     name: beacon-chain-tests
-    runs-on: ubuntu-latest
+    # Use self-hosted runners only on the sigp repo.
+    runs-on: ${{ github.repository == 'sigp/lighthouse' && fromJson('["self-hosted", "linux", "large"]') || 'ubuntu-latest'  }}
     needs: cargo-fmt
     steps:
     - uses: actions/checkout@v3
     - name: Get latest version of stable Rust
+      if: env.SELF_HOSTED_RUNNERS == false
       run: rustup update stable
     - name: Install Protoc
       uses: arduino/setup-protoc@e52d9eb8f7b63115df1ac544a1376fdbf5a39612
@@ -144,11 +154,13 @@ jobs:
       run: make test-slasher
   debug-tests-ubuntu:
     name: debug-tests-ubuntu
-    runs-on: ubuntu-22.04
+    # Use self-hosted runners only on the sigp repo.
+    runs-on: ${{ github.repository == 'sigp/lighthouse' && fromJson('["self-hosted", "linux", "large"]') || 'ubuntu-latest'  }}
     needs: cargo-fmt
     steps:
     - uses: actions/checkout@v3
     - name: Get latest version of stable Rust
+      if: env.SELF_HOSTED_RUNNERS == false
       run: rustup update stable
     - name: Install Protoc
       uses: arduino/setup-protoc@e52d9eb8f7b63115df1ac544a1376fdbf5a39612
@@ -174,11 +186,13 @@ jobs:
       run: make run-state-transition-tests
   ef-tests-ubuntu:
     name: ef-tests-ubuntu
-    runs-on: ubuntu-latest
+    # Use self-hosted runners only on the sigp repo.
+    runs-on: ${{ github.repository == 'sigp/lighthouse' && fromJson('["self-hosted", "linux", "small"]') || 'ubuntu-latest'  }}
     needs: cargo-fmt
     steps:
     - uses: actions/checkout@v3
     - name: Get latest version of stable Rust
+      if: env.SELF_HOSTED_RUNNERS == false
       run: rustup update stable
     - name: Install Protoc
       uses: arduino/setup-protoc@e52d9eb8f7b63115df1ac544a1376fdbf5a39612

Cargo.lock

@@ -643,6 +643,7 @@ dependencies = [
  "eth2",
  "eth2_network_config",
  "ethereum_hashing",
+ "ethereum_serde_utils",
  "ethereum_ssz",
  "ethereum_ssz_derive",
  "execution_layer",

beacon_node/beacon_chain/Cargo.toml

@@ -29,6 +29,7 @@ operation_pool = { path = "../operation_pool" }
 rayon = "1.4.1"
 serde = "1.0.116"
 serde_derive = "1.0.116"
+ethereum_serde_utils = "0.5.0"
 slog = { version = "2.5.2", features = ["max_level_trace"] }
 sloggers = { version = "2.1.1", features = ["json"] }
 slot_clock = { path = "../../common/slot_clock" }

beacon_node/beacon_chain/src/attestation_rewards.rs

@@ -3,7 +3,8 @@ use eth2::lighthouse::attestation_rewards::{IdealAttestationRewards, TotalAttest
 use eth2::lighthouse::StandardAttestationRewards;
 use participation_cache::ParticipationCache;
 use safe_arith::SafeArith;
-use slog::{debug, Logger};
+use serde_utils::quoted_u64::Quoted;
+use slog::debug;
 use state_processing::{
     common::altair::BaseRewardPerIncrement,
     per_epoch_processing::altair::{participation_cache, rewards_and_penalties::get_flag_weight},
@@ -15,32 +16,111 @@ use store::consts::altair::{
 };
 use types::consts::altair::WEIGHT_DENOMINATOR;
 
-use types::{Epoch, EthSpec};
+use types::{BeaconState, Epoch, EthSpec};
 
 use eth2::types::ValidatorId;
+use state_processing::common::base::get_base_reward_from_effective_balance;
+use state_processing::per_epoch_processing::base::rewards_and_penalties::{
+    get_attestation_component_delta, get_attestation_deltas_all, get_attestation_deltas_subset,
+    get_inactivity_penalty_delta, get_inclusion_delay_delta,
+};
+use state_processing::per_epoch_processing::base::validator_statuses::InclusionInfo;
+use state_processing::per_epoch_processing::base::{
+    TotalBalances, ValidatorStatus, ValidatorStatuses,
+};
 
 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());
+        debug!(self.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
+        let state = self
             .get_state(&state_root, Some(state_slot))?
             .ok_or(BeaconChainError::MissingBeaconState(state_root))?;
 
+        match state {
+            BeaconState::Base(_) => self.compute_attestation_rewards_base(state, validators),
+            BeaconState::Altair(_) | BeaconState::Merge(_) | BeaconState::Capella(_) => {
+                self.compute_attestation_rewards_altair(state, validators)
+            }
+        }
+    }
+
+    fn compute_attestation_rewards_base(
+        &self,
+        mut state: BeaconState<T::EthSpec>,
+        validators: Vec<ValidatorId>,
+    ) -> Result<StandardAttestationRewards, BeaconChainError> {
+        let spec = &self.spec;
+        let mut validator_statuses = ValidatorStatuses::new(&state, spec)?;
+        validator_statuses.process_attestations(&state)?;
+
+        let ideal_rewards =
+            self.compute_ideal_rewards_base(&state, &validator_statuses.total_balances)?;
+
+        let indices_to_attestation_delta = if validators.is_empty() {
+            get_attestation_deltas_all(&state, &validator_statuses, spec)?
+                .into_iter()
+                .enumerate()
+                .collect()
+        } else {
+            let validator_indices = Self::validators_ids_to_indices(&mut state, validators)?;
+            get_attestation_deltas_subset(&state, &validator_statuses, &validator_indices, spec)?
+        };
+
+        let mut total_rewards = vec![];
+
+        for (index, delta) in indices_to_attestation_delta.into_iter() {
+            let head_delta = delta.head_delta;
+            let head = (head_delta.rewards as i64).safe_sub(head_delta.penalties as i64)?;
+
+            let target_delta = delta.target_delta;
+            let target = (target_delta.rewards as i64).safe_sub(target_delta.penalties as i64)?;
+
+            let source_delta = delta.source_delta;
+            let source = (source_delta.rewards as i64).safe_sub(source_delta.penalties as i64)?;
+
+            // No penalties associated with inclusion delay
+            let inclusion_delay = delta.inclusion_delay_delta.rewards;
+            let inactivity = delta.inactivity_penalty_delta.penalties.wrapping_neg() as i64;
+
+            let rewards = TotalAttestationRewards {
+                validator_index: index as u64,
+                head,
+                target,
+                source,
+                inclusion_delay: Some(Quoted {
+                    value: inclusion_delay,
+                }),
+                inactivity,
+            };
+
+            total_rewards.push(rewards);
+        }
+
+        Ok(StandardAttestationRewards {
+            ideal_rewards,
+            total_rewards,
+        })
+    }
+
+    fn compute_attestation_rewards_altair(
+        &self,
+        mut state: BeaconState<T::EthSpec>,
+        validators: Vec<ValidatorId>,
+    ) -> Result<StandardAttestationRewards, BeaconChainError> {
+        let spec = &self.spec;
+
         // Calculate ideal_rewards
         let participation_cache = ParticipationCache::new(&state, spec)?;
 
@@ -71,7 +151,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             let base_reward_per_increment =
                 BaseRewardPerIncrement::new(total_active_balance, spec)?;
 
-            for effective_balance_eth in 0..=32 {
+            for effective_balance_eth in 1..=self.max_effective_balance_increment_steps()? {
                 let effective_balance =
                     effective_balance_eth.safe_mul(spec.effective_balance_increment)?;
                 let base_reward =
@@ -101,20 +181,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let validators = if validators.is_empty() {
             participation_cache.eligible_validator_indices().to_vec()
         } else {
-            validators
+            Self::validators_ids_to_indices(&mut state, 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 head_reward = 0i64;
             let mut target_reward = 0i64;
             let mut source_reward = 0i64;
 
@@ -132,7 +204,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         .map_err(|_| BeaconChainError::AttestationRewardsError)?;
                     if voted_correctly {
                         if flag_index == TIMELY_HEAD_FLAG_INDEX {
-                            head_reward += ideal_reward;
+                            head_reward += *ideal_reward as i64;
                         } else if flag_index == TIMELY_TARGET_FLAG_INDEX {
                             target_reward += *ideal_reward as i64;
                         } else if flag_index == TIMELY_SOURCE_FLAG_INDEX {
@@ -152,6 +224,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 head: head_reward,
                 target: target_reward,
                 source: source_reward,
+                inclusion_delay: None,
+                // TODO: altair calculation logic needs to be updated to include inactivity penalty
+                inactivity: 0,
             });
         }
 
@@ -173,6 +248,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                             head: 0,
                             target: 0,
                             source: 0,
+                            inclusion_delay: None,
+                            // TODO: altair calculation logic needs to be updated to include inactivity penalty
+                            inactivity: 0,
                         });
                     match *flag_index {
                         TIMELY_SOURCE_FLAG_INDEX => entry.source += ideal_reward,
@@ -192,4 +270,126 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             total_rewards,
         })
     }
+
+    fn max_effective_balance_increment_steps(&self) -> Result<u64, BeaconChainError> {
+        let spec = &self.spec;
+        let max_steps = spec
+            .max_effective_balance
+            .safe_div(spec.effective_balance_increment)?;
+        Ok(max_steps)
+    }
+
+    fn validators_ids_to_indices(
+        state: &mut BeaconState<T::EthSpec>,
+        validators: Vec<ValidatorId>,
+    ) -> Result<Vec<usize>, BeaconChainError> {
+        let indices = 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<_>, _>>()?;
+        Ok(indices)
+    }
+
+    fn compute_ideal_rewards_base(
+        &self,
+        state: &BeaconState<T::EthSpec>,
+        total_balances: &TotalBalances,
+    ) -> Result<Vec<IdealAttestationRewards>, BeaconChainError> {
+        let spec = &self.spec;
+        let previous_epoch = state.previous_epoch();
+        let finality_delay = previous_epoch
+            .safe_sub(state.finalized_checkpoint().epoch)?
+            .as_u64();
+
+        let ideal_validator_status = ValidatorStatus {
+            is_previous_epoch_attester: true,
+            is_slashed: false,
+            inclusion_info: Some(InclusionInfo {
+                delay: 1,
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+
+        let mut ideal_attestation_rewards_list = Vec::new();
+
+        for effective_balance_step in 1..=self.max_effective_balance_increment_steps()? {
+            let effective_balance =
+                effective_balance_step.safe_mul(spec.effective_balance_increment)?;
+            let base_reward = get_base_reward_from_effective_balance::<T::EthSpec>(
+                effective_balance,
+                total_balances.current_epoch(),
+                spec,
+            )?;
+
+            // compute ideal head rewards
+            let head = get_attestation_component_delta(
+                true,
+                total_balances.previous_epoch_attesters(),
+                total_balances,
+                base_reward,
+                finality_delay,
+                spec,
+            )?
+            .rewards;
+
+            // compute ideal target rewards
+            let target = get_attestation_component_delta(
+                true,
+                total_balances.previous_epoch_target_attesters(),
+                total_balances,
+                base_reward,
+                finality_delay,
+                spec,
+            )?
+            .rewards;
+
+            // compute ideal source rewards
+            let source = get_attestation_component_delta(
+                true,
+                total_balances.previous_epoch_head_attesters(),
+                total_balances,
+                base_reward,
+                finality_delay,
+                spec,
+            )?
+            .rewards;
+
+            // compute ideal inclusion delay rewards
+            let inclusion_delay =
+                get_inclusion_delay_delta(&ideal_validator_status, base_reward, spec)?
+                    .0
+                    .rewards;
+
+            // compute inactivity penalty
+            let inactivity = get_inactivity_penalty_delta(
+                &ideal_validator_status,
+                base_reward,
+                finality_delay,
+                spec,
+            )?
+            .penalties
+            .wrapping_neg() as i64;
+
+            let ideal_attestation_rewards = IdealAttestationRewards {
+                effective_balance,
+                head,
+                target,
+                source,
+                inclusion_delay: Some(Quoted {
+                    value: inclusion_delay,
+                }),
+                inactivity,
+            };
+
+            ideal_attestation_rewards_list.push(ideal_attestation_rewards);
+        }
+
+        Ok(ideal_attestation_rewards_list)
+    }
 }

beacon_node/beacon_chain/src/errors.rs

@@ -26,7 +26,7 @@ use state_processing::{
     },
     signature_sets::Error as SignatureSetError,
     state_advance::Error as StateAdvanceError,
-    BlockProcessingError, BlockReplayError, SlotProcessingError,
+    BlockProcessingError, BlockReplayError, EpochProcessingError, SlotProcessingError,
 };
 use std::time::Duration;
 use task_executor::ShutdownReason;
@@ -62,6 +62,7 @@ pub enum BeaconChainError {
     MissingBeaconBlock(Hash256),
     MissingBeaconState(Hash256),
     SlotProcessingError(SlotProcessingError),
+    EpochProcessingError(EpochProcessingError),
     StateAdvanceError(StateAdvanceError),
     UnableToAdvanceState(String),
     NoStateForAttestation {
@@ -221,6 +222,7 @@ pub enum BeaconChainError {
 }
 
 easy_from_to!(SlotProcessingError, BeaconChainError);
+easy_from_to!(EpochProcessingError, BeaconChainError);
 easy_from_to!(AttestationValidationError, BeaconChainError);
 easy_from_to!(SyncCommitteeMessageValidationError, BeaconChainError);
 easy_from_to!(ExitValidationError, BeaconChainError);

beacon_node/beacon_chain/tests/rewards.rs

@@ -9,19 +9,22 @@ use beacon_chain::{
     test_utils::{AttestationStrategy, BlockStrategy, RelativeSyncCommittee},
     types::{Epoch, EthSpec, Keypair, MinimalEthSpec},
 };
+use eth2::lighthouse::attestation_rewards::TotalAttestationRewards;
+use eth2::lighthouse::StandardAttestationRewards;
+use eth2::types::ValidatorId;
 use lazy_static::lazy_static;
+use types::beacon_state::Error as BeaconStateError;
+use types::{BeaconState, ChainSpec};
 
 pub const VALIDATOR_COUNT: usize = 64;
 
+type E = MinimalEthSpec;
+
 lazy_static! {
     static ref KEYPAIRS: Vec<Keypair> = generate_deterministic_keypairs(VALIDATOR_COUNT);
 }
 
-fn get_harness<E: EthSpec>() -> BeaconChainHarness<EphemeralHarnessType<E>> {
+fn get_harness(spec: ChainSpec) -> 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())
@@ -35,8 +38,11 @@ fn get_harness<E: EthSpec>() -> BeaconChainHarness<EphemeralHarnessType<E>> {
 
 #[tokio::test]
 async fn test_sync_committee_rewards() {
-    let num_block_produced = MinimalEthSpec::slots_per_epoch();
+    let mut spec = E::default_spec();
-    let harness = get_harness::<MinimalEthSpec>();
+    spec.altair_fork_epoch = Some(Epoch::new(0));
+
+    let harness = get_harness(spec);
+    let num_block_produced = E::slots_per_epoch();
 
     let latest_block_root = harness
         .extend_chain(
@@ -119,3 +125,175 @@ async fn test_sync_committee_rewards() {
         mismatches.join(",")
     );
 }
+
+#[tokio::test]
+async fn test_verify_attestation_rewards_base() {
+    let harness = get_harness(E::default_spec());
+
+    // epoch 0 (N), only two thirds of validators vote.
+    let two_thirds = (VALIDATOR_COUNT / 3) * 2;
+    let two_thirds_validators: Vec<usize> = (0..two_thirds).collect();
+    harness
+        .extend_chain(
+            E::slots_per_epoch() as usize,
+            BlockStrategy::OnCanonicalHead,
+            AttestationStrategy::SomeValidators(two_thirds_validators),
+        )
+        .await;
+
+    let initial_balances: Vec<u64> = harness.get_current_state().balances().clone().into();
+
+    // extend slots to beginning of epoch N + 2
+    harness.extend_slots(E::slots_per_epoch() as usize).await;
+
+    // compute reward deltas for all validators in epoch N
+    let StandardAttestationRewards {
+        ideal_rewards,
+        total_rewards,
+    } = harness
+        .chain
+        .compute_attestation_rewards(Epoch::new(0), vec![])
+        .unwrap();
+
+    // assert no inactivity penalty for both ideal rewards and individual validators
+    assert!(ideal_rewards.iter().all(|reward| reward.inactivity == 0));
+    assert!(total_rewards.iter().all(|reward| reward.inactivity == 0));
+
+    // apply attestation rewards to initial balances
+    let expected_balances = apply_attestation_rewards(&initial_balances, total_rewards);
+
+    // verify expected balances against actual balances
+    let balances: Vec<u64> = harness.get_current_state().balances().clone().into();
+    assert_eq!(expected_balances, balances);
+}
+
+#[tokio::test]
+async fn test_verify_attestation_rewards_base_inactivity_leak() {
+    let spec = E::default_spec();
+    let harness = get_harness(spec.clone());
+
+    let half = VALIDATOR_COUNT / 2;
+    let half_validators: Vec<usize> = (0..half).collect();
+    // target epoch is the epoch where the chain enters inactivity leak
+    let target_epoch = &spec.min_epochs_to_inactivity_penalty + 1;
+
+    // advance until beginning of epoch N + 1 and get balances
+    harness
+        .extend_chain(
+            (E::slots_per_epoch() * (target_epoch + 1)) as usize,
+            BlockStrategy::OnCanonicalHead,
+            AttestationStrategy::SomeValidators(half_validators.clone()),
+        )
+        .await;
+    let initial_balances: Vec<u64> = harness.get_current_state().balances().clone().into();
+
+    // extend slots to beginning of epoch N + 2
+    harness.advance_slot();
+    harness
+        .extend_chain(
+            E::slots_per_epoch() as usize,
+            BlockStrategy::OnCanonicalHead,
+            AttestationStrategy::SomeValidators(half_validators),
+        )
+        .await;
+    let _slot = harness.get_current_slot();
+
+    // compute reward deltas for all validators in epoch N
+    let StandardAttestationRewards {
+        ideal_rewards,
+        total_rewards,
+    } = harness
+        .chain
+        .compute_attestation_rewards(Epoch::new(target_epoch), vec![])
+        .unwrap();
+
+    // assert inactivity penalty for both ideal rewards and individual validators
+    assert!(ideal_rewards.iter().all(|reward| reward.inactivity < 0));
+    assert!(total_rewards.iter().all(|reward| reward.inactivity < 0));
+
+    // apply attestation rewards to initial balances
+    let expected_balances = apply_attestation_rewards(&initial_balances, total_rewards);
+
+    // verify expected balances against actual balances
+    let balances: Vec<u64> = harness.get_current_state().balances().clone().into();
+    assert_eq!(expected_balances, balances);
+}
+
+#[tokio::test]
+async fn test_verify_attestation_rewards_base_subset_only() {
+    let harness = get_harness(E::default_spec());
+
+    // epoch 0 (N), only two thirds of validators vote.
+    let two_thirds = (VALIDATOR_COUNT / 3) * 2;
+    let two_thirds_validators: Vec<usize> = (0..two_thirds).collect();
+    harness
+        .extend_chain(
+            E::slots_per_epoch() as usize,
+            BlockStrategy::OnCanonicalHead,
+            AttestationStrategy::SomeValidators(two_thirds_validators),
+        )
+        .await;
+
+    // a small subset of validators to compute attestation rewards for
+    let validators_subset = [0, VALIDATOR_COUNT / 2, VALIDATOR_COUNT - 1];
+
+    // capture balances before transitioning to N + 2
+    let initial_balances = get_validator_balances(harness.get_current_state(), &validators_subset);
+
+    // extend slots to beginning of epoch N + 2
+    harness.extend_slots(E::slots_per_epoch() as usize).await;
+
+    let validators_subset_ids: Vec<ValidatorId> = validators_subset
+        .into_iter()
+        .map(|idx| ValidatorId::Index(idx as u64))
+        .collect();
+
+    // compute reward deltas for the subset of validators in epoch N
+    let StandardAttestationRewards {
+        ideal_rewards: _,
+        total_rewards,
+    } = harness
+        .chain
+        .compute_attestation_rewards(Epoch::new(0), validators_subset_ids)
+        .unwrap();
+
+    // apply attestation rewards to initial balances
+    let expected_balances = apply_attestation_rewards(&initial_balances, total_rewards);
+
+    // verify expected balances against actual balances
+    let balances = get_validator_balances(harness.get_current_state(), &validators_subset);
+    assert_eq!(expected_balances, balances);
+}
+
+/// Apply a vec of `TotalAttestationRewards` to initial balances, and return
+fn apply_attestation_rewards(
+    initial_balances: &[u64],
+    attestation_rewards: Vec<TotalAttestationRewards>,
+) -> Vec<u64> {
+    initial_balances
+        .iter()
+        .zip(attestation_rewards)
+        .map(|(&initial_balance, rewards)| {
+            let expected_balance = initial_balance as i64
+                + rewards.head
+                + rewards.source
+                + rewards.target
+                + rewards.inclusion_delay.map(|q| q.value).unwrap_or(0) as i64
+                + rewards.inactivity;
+            expected_balance as u64
+        })
+        .collect::<Vec<u64>>()
+}
+
+fn get_validator_balances(state: BeaconState<E>, validators: &[usize]) -> Vec<u64> {
+    validators
+        .iter()
+        .flat_map(|&id| {
+            state
+                .balances()
+                .get(id)
+                .cloned()
+                .ok_or(BeaconStateError::BalancesOutOfBounds(id))
+        })
+        .collect()
+}

beacon_node/http_api/src/lib.rs

@@ -2075,15 +2075,11 @@ pub fn serve<T: BeaconChainTypes>(
         .and(warp::path::param::<Epoch>())
         .and(warp::path::end())
         .and(warp::body::json())
-        .and(log_filter.clone())
         .and_then(
-            |chain: Arc<BeaconChain<T>>,
+            |chain: Arc<BeaconChain<T>>, epoch: Epoch, validators: Vec<ValidatorId>| {
-             epoch: Epoch,
-             validators: Vec<ValidatorId>,
-             log: Logger| {
                 blocking_json_task(move || {
                     let attestation_rewards = chain
-                        .compute_attestation_rewards(epoch, validators, log)
+                        .compute_attestation_rewards(epoch, validators)
                         .map_err(|e| match e {
                             BeaconChainError::MissingBeaconState(root) => {
                                 warp_utils::reject::custom_not_found(format!(

common/eth2/src/lighthouse/attestation_rewards.rs

@@ -1,4 +1,5 @@
 use serde::{Deserialize, Serialize};
+use serde_utils::quoted_u64::Quoted;
 
 // Details about the rewards paid for attestations
 // All rewards in GWei
@@ -17,6 +18,12 @@ pub struct IdealAttestationRewards {
     // Ideal attester's reward for source vote in gwei
     #[serde(with = "serde_utils::quoted_u64")]
     pub source: u64,
+    // Ideal attester's inclusion_delay reward in gwei (phase0 only)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub inclusion_delay: Option<Quoted<u64>>,
+    // Ideal attester's inactivity penalty in gwei
+    #[serde(with = "serde_utils::quoted_i64")]
+    pub inactivity: i64,
 }
 
 #[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
@@ -25,16 +32,20 @@ pub struct TotalAttestationRewards {
     #[serde(with = "serde_utils::quoted_u64")]
     pub validator_index: u64,
     // attester's reward for head vote in gwei
-    #[serde(with = "serde_utils::quoted_u64")]
+    #[serde(with = "serde_utils::quoted_i64")]
-    pub head: u64,
+    pub head: i64,
     // attester's reward for target vote in gwei
     #[serde(with = "serde_utils::quoted_i64")]
     pub target: i64,
     // attester's reward for source vote in gwei
     #[serde(with = "serde_utils::quoted_i64")]
     pub source: i64,
-    // TBD attester's inclusion_delay reward in gwei (phase0 only)
+    // attester's inclusion_delay reward in gwei (phase0 only)
-    // pub inclusion_delay: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub inclusion_delay: Option<Quoted<u64>>,
+    // attester's inactivity penalty in gwei
+    #[serde(with = "serde_utils::quoted_i64")]
+    pub inactivity: i64,
 }
 
 #[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]

consensus/state_processing/src/common/base.rs

@@ -17,3 +17,15 @@ pub fn get_base_reward<T: EthSpec>(
         .safe_div(spec.base_rewards_per_epoch)
         .map_err(Into::into)
 }
+
+pub fn get_base_reward_from_effective_balance<T: EthSpec>(
+    effective_balance: u64,
+    total_active_balance: u64,
+    spec: &ChainSpec,
+) -> Result<u64, BeaconStateError> {
+    effective_balance
+        .safe_mul(spec.base_reward_factor)?
+        .safe_div(total_active_balance.integer_sqrt())?
+        .safe_div(spec.base_rewards_per_epoch)
+        .map_err(Into::into)
+}

consensus/state_processing/src/per_epoch_processing/base.rs

@@ -36,7 +36,7 @@ pub fn process_epoch<T: EthSpec>(
     justification_and_finalization_state.apply_changes_to_state(state);
 
     // Rewards and Penalties.
-    process_rewards_and_penalties(state, &mut validator_statuses, spec)?;
+    process_rewards_and_penalties(state, &validator_statuses, spec)?;
 
     // Registry Updates.
     process_registry_updates(state, spec)?;

consensus/state_processing/src/per_epoch_processing/base/rewards_and_penalties.rs

@@ -45,7 +45,7 @@ impl AttestationDelta {
 /// Apply attester and proposer rewards.
 pub fn process_rewards_and_penalties<T: EthSpec>(
     state: &mut BeaconState<T>,
-    validator_statuses: &mut ValidatorStatuses,
+    validator_statuses: &ValidatorStatuses,
     spec: &ChainSpec,
 ) -> Result<(), Error> {
     if state.current_epoch() == T::genesis_epoch() {
@@ -59,7 +59,7 @@ pub fn process_rewards_and_penalties<T: EthSpec>(
         return Err(Error::ValidatorStatusesInconsistent);
     }
 
-    let deltas = get_attestation_deltas(state, validator_statuses, spec)?;
+    let deltas = get_attestation_deltas_all(state, validator_statuses, spec)?;
 
     // Apply the deltas, erroring on overflow above but not on overflow below (saturating at 0
     // instead).
@@ -73,10 +73,41 @@ pub fn process_rewards_and_penalties<T: EthSpec>(
 }
 
 /// Apply rewards for participation in attestations during the previous epoch.
-pub fn get_attestation_deltas<T: EthSpec>(
+pub fn get_attestation_deltas_all<T: EthSpec>(
     state: &BeaconState<T>,
     validator_statuses: &ValidatorStatuses,
     spec: &ChainSpec,
+) -> Result<Vec<AttestationDelta>, Error> {
+    get_attestation_deltas(state, validator_statuses, None, spec)
+}
+
+/// Apply rewards for participation in attestations during the previous epoch, and only compute
+/// rewards for a subset of validators.
+pub fn get_attestation_deltas_subset<T: EthSpec>(
+    state: &BeaconState<T>,
+    validator_statuses: &ValidatorStatuses,
+    validators_subset: &Vec<usize>,
+    spec: &ChainSpec,
+) -> Result<Vec<(usize, AttestationDelta)>, Error> {
+    get_attestation_deltas(state, validator_statuses, Some(validators_subset), spec).map(|deltas| {
+        deltas
+            .into_iter()
+            .enumerate()
+            .filter(|(index, _)| validators_subset.contains(index))
+            .collect()
+    })
+}
+
+/// Apply rewards for participation in attestations during the previous epoch.
+/// If `maybe_validators_subset` specified, only the deltas for the specified validator subset is
+/// returned, otherwise deltas for all validators are returned.
+///
+/// Returns a vec of validator indices to `AttestationDelta`.
+fn get_attestation_deltas<T: EthSpec>(
+    state: &BeaconState<T>,
+    validator_statuses: &ValidatorStatuses,
+    maybe_validators_subset: Option<&Vec<usize>>,
+    spec: &ChainSpec,
 ) -> Result<Vec<AttestationDelta>, Error> {
     let previous_epoch = state.previous_epoch();
     let finality_delay = state
@@ -88,6 +119,13 @@ pub fn get_attestation_deltas<T: EthSpec>(
 
     let total_balances = &validator_statuses.total_balances;
 
+    // Ignore validator if a subset is specified and validator is not in the subset
+    let include_validator_delta = |idx| match maybe_validators_subset.as_ref() {
+        None => true,
+        Some(validators_subset) if validators_subset.contains(&idx) => true,
+        Some(_) => false,
+    };
+
     for (index, validator) in validator_statuses.statuses.iter().enumerate() {
         // Ignore ineligible validators. All sub-functions of the spec do this except for
         // `get_inclusion_delay_deltas`. It's safe to do so here because any validator that is in
@@ -99,41 +137,46 @@ pub fn get_attestation_deltas<T: EthSpec>(
 
         let base_reward = get_base_reward(state, index, total_balances.current_epoch(), spec)?;
 
-        let source_delta =
-            get_source_delta(validator, base_reward, total_balances, finality_delay, spec)?;
-        let target_delta =
-            get_target_delta(validator, base_reward, total_balances, finality_delay, spec)?;
-        let head_delta =
-            get_head_delta(validator, base_reward, total_balances, finality_delay, spec)?;
         let (inclusion_delay_delta, proposer_delta) =
             get_inclusion_delay_delta(validator, base_reward, spec)?;
-        let inactivity_penalty_delta =
-            get_inactivity_penalty_delta(validator, base_reward, finality_delay, spec)?;
 
-        let delta = deltas
+        if include_validator_delta(index) {
-            .get_mut(index)
+            let source_delta =
-            .ok_or(Error::DeltaOutOfBounds(index))?;
+                get_source_delta(validator, base_reward, total_balances, finality_delay, spec)?;
-        delta.source_delta.combine(source_delta)?;
+            let target_delta =
-        delta.target_delta.combine(target_delta)?;
+                get_target_delta(validator, base_reward, total_balances, finality_delay, spec)?;
-        delta.head_delta.combine(head_delta)?;
+            let head_delta =
-        delta.inclusion_delay_delta.combine(inclusion_delay_delta)?;
+                get_head_delta(validator, base_reward, total_balances, finality_delay, spec)?;
-        delta
+            let inactivity_penalty_delta =
-            .inactivity_penalty_delta
+                get_inactivity_penalty_delta(validator, base_reward, finality_delay, spec)?;
-            .combine(inactivity_penalty_delta)?;
+
+            let delta = deltas
+                .get_mut(index)
+                .ok_or(Error::DeltaOutOfBounds(index))?;
+            delta.source_delta.combine(source_delta)?;
+            delta.target_delta.combine(target_delta)?;
+            delta.head_delta.combine(head_delta)?;
+            delta.inclusion_delay_delta.combine(inclusion_delay_delta)?;
+            delta
+                .inactivity_penalty_delta
+                .combine(inactivity_penalty_delta)?;
+        }
 
         if let Some((proposer_index, proposer_delta)) = proposer_delta {
-            deltas
+            if include_validator_delta(proposer_index) {
-                .get_mut(proposer_index)
+                deltas
-                .ok_or(Error::ValidatorStatusesInconsistent)?
+                    .get_mut(proposer_index)
-                .inclusion_delay_delta
+                    .ok_or(Error::ValidatorStatusesInconsistent)?
-                .combine(proposer_delta)?;
+                    .inclusion_delay_delta
+                    .combine(proposer_delta)?;
+            }
         }
     }
 
     Ok(deltas)
 }
 
-fn get_attestation_component_delta(
+pub fn get_attestation_component_delta(
     index_in_unslashed_attesting_indices: bool,
     attesting_balance: u64,
     total_balances: &TotalBalances,
@@ -216,7 +259,7 @@ fn get_head_delta(
     )
 }
 
-fn get_inclusion_delay_delta(
+pub fn get_inclusion_delay_delta(
     validator: &ValidatorStatus,
     base_reward: u64,
     spec: &ChainSpec,
@@ -242,7 +285,7 @@ fn get_inclusion_delay_delta(
     }
 }
 
-fn get_inactivity_penalty_delta(
+pub fn get_inactivity_penalty_delta(
     validator: &ValidatorStatus,
     base_reward: u64,
     finality_delay: u64,

testing/ef_tests/src/cases/epoch_processing.rs

@@ -123,7 +123,7 @@ impl<E: EthSpec> EpochTransition<E> for RewardsAndPenalties {
             BeaconState::Base(_) => {
                 let mut validator_statuses = base::ValidatorStatuses::new(state, spec)?;
                 validator_statuses.process_attestations(state)?;
-                base::process_rewards_and_penalties(state, &mut validator_statuses, spec)
+                base::process_rewards_and_penalties(state, &validator_statuses, spec)
             }
             BeaconState::Altair(_)
             | BeaconState::Merge(_)

testing/ef_tests/src/cases/rewards.rs

@@ -118,7 +118,7 @@ impl<E: EthSpec> Case for RewardsTest<E> {
                 let mut validator_statuses = ValidatorStatuses::new(&state, spec)?;
                 validator_statuses.process_attestations(&state)?;
 
-                let deltas = base::rewards_and_penalties::get_attestation_deltas(
+                let deltas = base::rewards_and_penalties::get_attestation_deltas_all(
                     &state,
                     &validator_statuses,
                     spec,

watch/tests/tests.rs

@@ -22,6 +22,7 @@ use watch::{
 };
 
 use log::error;
+use std::env;
 use std::net::SocketAddr;
 use std::time::Duration;
 use tokio::{runtime, task::JoinHandle};
@@ -36,6 +37,11 @@ const VALIDATOR_COUNT: usize = 32;
 const SLOTS_PER_EPOCH: u64 = 32;
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
 
+/// Set this environment variable to use a different hostname for connecting to
+/// the database. Can be set to `host.docker.internal` for docker-in-docker
+/// setups.
+const WATCH_HOST_ENV_VARIABLE: &str = "WATCH_HOST";
+
 fn build_test_config(config: &DatabaseConfig) -> PostgresConfig {
     let mut postgres_config = PostgresConfig::new();
     postgres_config
@@ -71,6 +77,10 @@ pub async fn create_test_database(config: &DatabaseConfig) {
         .expect("Database creation failed");
 }
 
+pub fn get_host_from_env() -> String {
+    env::var(WATCH_HOST_ENV_VARIABLE).unwrap_or_else(|_| "localhost".to_string())
+}
+
 struct TesterBuilder {
     pub harness: BeaconChainHarness<EphemeralHarnessType<E>>,
     pub config: Config,
@@ -107,6 +117,7 @@ impl TesterBuilder {
             database: DatabaseConfig {
                 dbname: random_dbname(),
                 port: database_port,
+                host: get_host_from_env(),
                 ..Default::default()
             },
             server: ServerConfig {