add consolidation processing

consensus/state_processing/src/per_block_processing/block_signature_verifier.rs

@@ -171,6 +171,7 @@ where
         self.include_exits(block)?;
         self.include_sync_aggregate(block)?;
         self.include_bls_to_execution_changes(block)?;
+        self.include_consolidations(block)?;
 
         Ok(())
     }
@@ -359,6 +360,27 @@ where
         Ok(())
     }
 
+    /// Includes all signatures in `self.block.body.consolidations` for verification.
+    pub fn include_consolidations<Payload: AbstractExecPayload<E>>(
+        &mut self,
+        block: &'a SignedBeaconBlock<E, Payload>,
+    ) -> Result<()> {
+        if let Ok(consolidations) = block.message().body().consolidations() {
+            self.sets.sets.reserve(consolidations.len());
+            for consolidation in consolidations {
+                let set = consolidation_signature_set(
+                    self.state,
+                    self.get_pubkey.clone(),
+                    consolidation,
+                    self.spec,
+                )?;
+
+                self.sets.push(set);
+            }
+        }
+        Ok(())
+    }
+
     /// Verify all the signatures that have been included in `self`, returning `true` if and only if
     /// all the signatures are valid.
     ///

consensus/state_processing/src/per_block_processing/errors.rs

@@ -89,6 +89,46 @@ pub enum BlockProcessingError {
         found: Hash256,
     },
     WithdrawalCredentialsInvalid,
+    TooManyPendingConsolidations {
+        consolidations: usize,
+        limit: usize,
+    },
+    ConsolidationChurnLimitTooLow {
+        churn_limit: u64,
+        minimum: u64,
+    },
+    MatchingSourceTargetConsolidation {
+        index: u64,
+    },
+    InactiveConsolidationSource {
+        index: u64,
+        current_epoch: Epoch,
+    },
+    InactiveConsolidationTarget {
+        index: u64,
+        current_epoch: Epoch,
+    },
+    SourceValidatorExiting {
+        index: u64,
+    },
+    TargetValidatorExiting {
+        index: u64,
+    },
+    FutureConsolidationEpoch {
+        current_epoch: Epoch,
+        consolidation_epoch: Epoch,
+    },
+    NoSourceExecutionWithdrawalCredential {
+        index: u64,
+    },
+    NoTargetExecutionWithdrawalCredential {
+        index: u64,
+    },
+    MismatchedWithdrawalCredentials {
+        source_address: Address,
+        target_address: Address,
+    },
+    InavlidConsolidationSignature,
 }
 
 impl From<BeaconStateError> for BlockProcessingError {

consensus/state_processing/src/per_block_processing/process_operations.rs

@@ -4,6 +4,7 @@ use crate::common::{
     slash_validator,
 };
 use crate::per_block_processing::errors::{BlockProcessingError, IntoWithIndex};
+use crate::signature_sets::consolidation_signature_set;
 use crate::VerifySignatures;
 use types::consts::altair::{PARTICIPATION_FLAG_WEIGHTS, PROPOSER_WEIGHT, WEIGHT_DENOMINATOR};
 use types::typenum::U33;
@@ -640,3 +641,161 @@ pub fn process_deposit_receipts<E: EthSpec>(
 
     Ok(())
 }
+
+pub fn process_consolidations<E: EthSpec>(
+    state: &mut BeaconState<E>,
+    consolidations: &[SignedConsolidation],
+    verify_signatures: VerifySignatures,
+    spec: &ChainSpec,
+) -> Result<(), BlockProcessingError> {
+    // If the pending consolidations queue is full, no consolidations are allowed in the block
+    let pending_consolidations = state.pending_consolidations()?.len();
+    let pending_consolidations_limit = E::pending_consolidations_limit();
+    block_verify! {
+        pending_consolidations < pending_consolidations_limit,
+        BlockProcessingError::TooManyPendingConsolidations {
+            consolidations: pending_consolidations,
+            limit: pending_consolidations_limit
+        }
+    }
+
+    // If there is too little available consolidation churn limit, no consolidations are allowed in the block
+    let churn_limit = state.get_consolidation_churn_limit(spec)?;
+    block_verify! {
+        churn_limit > spec.min_activation_balance,
+        BlockProcessingError::ConsolidationChurnLimitTooLow {
+            churn_limit,
+            minimum: spec.min_activation_balance
+        }
+    }
+
+    for signed_consolidation in consolidations {
+        let consolidation = signed_consolidation.message.clone();
+
+        // Verify that source != target, so a consolidation cannot be used as an exit.
+        block_verify! {
+            consolidation.source_index != consolidation.target_index,
+            BlockProcessingError::MatchingSourceTargetConsolidation  {
+                index: consolidation.source_index
+            }
+        }
+
+        let source_validator = state
+            .validators()
+            .get(consolidation.source_index as usize)
+            .ok_or(BeaconStateError::UnknownValidator(
+                consolidation.source_index as usize,
+            ))?;
+        let target_validator = state
+            .validators()
+            .get(consolidation.target_index as usize)
+            .ok_or(BeaconStateError::UnknownValidator(
+                consolidation.target_index as usize,
+            ))?;
+
+        // Verify the source and the target are active
+        let current_epoch = state.current_epoch();
+        block_verify! {
+            source_validator.is_active_at(current_epoch),
+            BlockProcessingError::InactiveConsolidationSource{
+                index: consolidation.source_index,
+                current_epoch
+            }
+        }
+        block_verify! {
+            target_validator.is_active_at(current_epoch),
+            BlockProcessingError::InactiveConsolidationTarget{
+                index: consolidation.target_index,
+                current_epoch
+            }
+        }
+
+        // Verify exits for source and target have not been initiated
+        block_verify! {
+            source_validator.exit_epoch == spec.far_future_epoch,
+            BlockProcessingError::SourceValidatorExiting{
+                index: consolidation.source_index,
+            }
+        }
+        block_verify! {
+            target_validator.exit_epoch == spec.far_future_epoch,
+            BlockProcessingError::TargetValidatorExiting{
+                index: consolidation.target_index,
+            }
+        }
+
+        // Consolidations must specify an epoch when they become valid; they are not valid before then
+        block_verify! {
+            current_epoch >= consolidation.epoch,
+            BlockProcessingError::FutureConsolidationEpoch {
+                current_epoch,
+                consolidation_epoch: consolidation.epoch
+            }
+        }
+
+        // Verify the source and the target have Execution layer withdrawal credentials
+        block_verify! {
+            source_validator.has_execution_withdrawal_credential(spec),
+            BlockProcessingError::NoSourceExecutionWithdrawalCredential {
+                index: consolidation.source_index,
+            }
+        }
+        block_verify! {
+            target_validator.has_execution_withdrawal_credential(spec),
+            BlockProcessingError::NoTargetExecutionWithdrawalCredential {
+                index: consolidation.target_index,
+            }
+        }
+
+        // Verify the same withdrawal address
+        let source_address = source_validator
+            .get_execution_withdrawal_address(spec)
+            .ok_or(BeaconStateError::NonExecutionAddresWithdrawalCredential)?;
+        let target_address = target_validator
+            .get_execution_withdrawal_address(spec)
+            .ok_or(BeaconStateError::NonExecutionAddresWithdrawalCredential)?;
+        block_verify! {
+            source_address == target_address,
+            BlockProcessingError::MismatchedWithdrawalCredentials {
+                source_address,
+                target_address
+            }
+        }
+
+        if verify_signatures.is_true() {
+            let signature_set = consolidation_signature_set(
+                state,
+                |i| get_pubkey_from_state(state, i),
+                signed_consolidation,
+                spec,
+            )?;
+            block_verify! {
+                signature_set.verify(),
+                BlockProcessingError::InavlidConsolidationSignature
+            }
+        }
+        let exit_epoch = state.compute_consolidation_epoch_and_update_churn(
+            source_validator.effective_balance,
+            spec,
+        )?;
+        let source_validator = state
+            .validators_mut()
+            .get_mut(consolidation.source_index as usize)
+            .ok_or(BeaconStateError::UnknownValidator(
+                consolidation.source_index as usize,
+            ))?;
+        // Initiate source validator exit and append pending consolidation
+        source_validator.exit_epoch = exit_epoch;
+        source_validator.withdrawable_epoch = source_validator
+            .exit_epoch
+            .safe_add(spec.min_validator_withdrawability_delay)?;
+        state
+            .pending_consolidations_mut()?
+            .push(PendingConsolidation {
+                source_index: consolidation.source_index,
+                target_index: consolidation.target_index,
+            })?;
+    }
+
+    Ok(())
+}

consensus/state_processing/src/per_block_processing/signature_sets.rs

@@ -11,8 +11,8 @@ use types::{
     BeaconStateError, ChainSpec, DepositData, Domain, Epoch, EthSpec, Fork, Hash256,
     InconsistentFork, IndexedAttestation, IndexedAttestationRef, ProposerSlashing, PublicKey,
     PublicKeyBytes, Signature, SignedAggregateAndProof, SignedBeaconBlock, SignedBeaconBlockHeader,
-    SignedBlsToExecutionChange, SignedContributionAndProof, SignedRoot, SignedVoluntaryExit,
+    SignedBlsToExecutionChange, SignedConsolidation, SignedContributionAndProof, SignedRoot,
-    SigningData, Slot, SyncAggregate, SyncAggregatorSelectionData, Unsigned,
+    SignedVoluntaryExit, SigningData, Slot, SyncAggregate, SyncAggregatorSelectionData, Unsigned,
 };
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -665,3 +665,38 @@ where
         message,
     )))
 }
+
+/// Returns two signature sets, one for the source and one for the target validator
+/// in the `SignedConsolidation`.
+pub fn consolidation_signature_set<'a, E, F>(
+    state: &'a BeaconState<E>,
+    get_pubkey: F,
+    consolidation: &'a SignedConsolidation,
+    spec: &'a ChainSpec,
+) -> Result<SignatureSet<'a>>
+where
+    E: EthSpec,
+    F: Fn(usize) -> Option<Cow<'a, PublicKey>>,
+{
+    let source_index = consolidation.message.source_index as usize;
+    let target_index = consolidation.message.target_index as usize;
+
+    let domain = spec.get_domain(
+        consolidation.message.epoch,
+        Domain::Consolidation,
+        &state.fork(),
+        state.genesis_validators_root(),
+    );
+
+    let message = consolidation.message.signing_root(domain);
+    let source_pubkey =
+        get_pubkey(source_index).ok_or(Error::ValidatorUnknown(source_index as u64))?;
+    let target_pubkey =
+        get_pubkey(target_index).ok_or(Error::ValidatorUnknown(target_index as u64))?;
+
+    Ok(SignatureSet::multiple_pubkeys(
+        &consolidation.signature,
+        vec![source_pubkey, target_pubkey],
+        message,
+    ))
+}

consensus/types/src/beacon_state.rs

@@ -160,6 +160,7 @@ pub enum Error {
     InvalidFlagIndex(usize),
     MerkleTreeError(merkle_proof::MerkleTreeError),
     PartialWithdrawalCountInvalid(usize),
+    NonExecutionAddresWithdrawalCredential,
 }
 
 /// Control whether an epoch-indexed field can be indexed at the next epoch or not.

consensus/types/src/consolidation.rs

@@ -1,5 +1,5 @@
-use crate::test_utils::TestRandom;
 use crate::Epoch;
+use crate::{test_utils::TestRandom, SignedRoot};
 use serde::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};
 use test_random_derive::TestRandom;
@@ -27,6 +27,8 @@ pub struct Consolidation {
     pub epoch: Epoch,
 }
 
+impl SignedRoot for Consolidation {}
+
 #[cfg(test)]
 mod tests {
     use super::*;

consensus/types/src/validator.rs

@@ -142,6 +142,18 @@ impl Validator {
             .flatten()
     }
 
+    /// Get the execution withdrawal address if this validator has one initialized.
+    pub fn get_execution_withdrawal_address(&self, spec: &ChainSpec) -> Option<Address> {
+        self.has_execution_withdrawal_credential(spec)
+            .then(|| {
+                self.withdrawal_credentials
+                    .as_bytes()
+                    .get(12..)
+                    .map(Address::from_slice)
+            })
+            .flatten()
+    }
+
     /// Changes withdrawal credentials to  the provided eth1 execution address.
     ///
     /// WARNING: this function does NO VALIDATION - it just does it!