Add op pool for consolidations

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -337,6 +337,7 @@ struct PartialBeaconBlock<E: EthSpec> {
     sync_aggregate: Option<SyncAggregate<E>>,
     prepare_payload_handle: Option<PreparePayloadHandle<E>>,
     bls_to_execution_changes: Vec<SignedBlsToExecutionChange>,
+    consolidations: Vec<SignedConsolidation>,
 }
 
 pub type BeaconForkChoice<T> = ForkChoice<
@@ -4885,6 +4886,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .op_pool
             .get_bls_to_execution_changes(&state, &self.spec);
 
+        let consolidations = self.op_pool.get_consolidations(&state, &self.spec);
+
         // Iterate through the naive aggregation pool and ensure all the attestations from there
         // are included in the operation pool.
         let unagg_import_timer =
@@ -5044,6 +5047,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             sync_aggregate,
             prepare_payload_handle,
             bls_to_execution_changes,
+            consolidations,
         })
     }
 
@@ -5072,6 +5076,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             // produce said `execution_payload`.
             prepare_payload_handle: _,
             bls_to_execution_changes,
+            consolidations,
         } = partial_beacon_block;
 
         let (inner_block, maybe_blobs_and_proofs, execution_payload_value) = match &state {
@@ -5215,7 +5220,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                                     "Kzg commitments missing from block contents".to_string(),
                                 ),
                             )?,
-                            consolidations: unimplemented!(),
+                            consolidations: consolidations.into(),
                         },
                     }),
                     maybe_blobs_and_proofs,

beacon_node/operation_pool/src/lib.rs

@@ -18,6 +18,7 @@ pub use persistence::{
     PersistedOperationPoolV15, PersistedOperationPoolV5,
 };
 pub use reward_cache::RewardCache;
+use types::SignedConsolidation;
 
 use crate::attestation_storage::{AttestationMap, CheckpointKey};
 use crate::bls_to_execution_changes::BlsToExecutionChanges;
@@ -29,7 +30,7 @@ use rand::seq::SliceRandom;
 use rand::thread_rng;
 use state_processing::per_block_processing::errors::AttestationValidationError;
 use state_processing::per_block_processing::{
-    get_slashable_indices_modular, verify_exit, VerifySignatures,
+    get_slashable_indices_modular, verify_consolidation, verify_exit, VerifySignatures,
 };
 use state_processing::{SigVerifiedOp, VerifyOperation};
 use std::collections::{hash_map::Entry, HashMap, HashSet};
@@ -58,6 +59,8 @@ pub struct OperationPool<T: EthSpec + Default> {
     voluntary_exits: RwLock<HashMap<u64, SigVerifiedOp<SignedVoluntaryExit, T>>>,
     /// Map from credential changing validator to their position in the queue.
     bls_to_execution_changes: RwLock<BlsToExecutionChanges<T>>,
+    /// Map from source validator index to consolidation.
+    consolidations: RwLock<HashMap<u64, SigVerifiedOp<SignedConsolidation, T>>>,
     /// Reward cache for accelerating attestation packing.
     reward_cache: RwLock<RewardCache>,
     _phantom: PhantomData<T>,
@@ -615,6 +618,44 @@ impl<T: EthSpec> OperationPool<T> {
             .prune(head_block, head_state, spec)
     }
 
+    /// Insert a consolidation into the pool.
+    pub fn insert_consolidation(
+        &self,
+        verified_consolidation: SigVerifiedOp<SignedConsolidation, T>,
+    ) {
+        self.consolidations.write().insert(
+            verified_consolidation.as_inner().message.source_index,
+            verified_consolidation,
+        );
+    }
+
+    /// Get a list of consolidations for inclusion in a block.
+    pub fn get_consolidations(
+        &self,
+        state: &BeaconState<T>,
+        spec: &ChainSpec,
+    ) -> Vec<SignedConsolidation> {
+        filter_limit_operations(
+            self.consolidations.read().values(),
+            |consolidation| {
+                consolidation.signature_is_still_valid(&state.fork())
+                // TODO(maxeb): do a more succint check of validity
+                    && verify_consolidation(state, consolidation.as_inner(), VerifySignatures::False, spec).is_ok()
+            },
+            |consolidation| consolidation.as_inner().clone(),
+            T::MaxConsolidations::to_usize(),
+        )
+    }
+
+    /// Prune consolidations for validators which are exited in the finalized epoch.
+    pub fn prune_consolidations(&self, head_state: &BeaconState<T>) {
+        prune_validator_hash_map(
+            &mut self.consolidations.write(),
+            |_, validator| validator.exit_epoch <= head_state.finalized_checkpoint().epoch,
+            head_state,
+        );
+    }
+
     /// Prune all types of transactions given the latest head state and head fork.
     pub fn prune_all<Payload: AbstractExecPayload<T>>(
         &self,
@@ -629,6 +670,7 @@ impl<T: EthSpec> OperationPool<T> {
         self.prune_attester_slashings(head_state);
         self.prune_voluntary_exits(head_state);
         self.prune_bls_to_execution_changes(head_block, head_state, spec);
+        self.prune_consolidations(head_state);
     }
 
     /// Total number of voluntary exits in the pool.
@@ -705,6 +747,17 @@ impl<T: EthSpec> OperationPool<T> {
             .map(|address_change| address_change.as_inner().clone())
             .collect()
     }
+
+    /// Returns all known `SignedConsolidation` objects.
+    ///
+    /// This method may return objects that are invalid for block inclusion.
+    pub fn get_all_consolidations(&self) -> Vec<SignedConsolidation> {
+        self.consolidations
+            .read()
+            .iter()
+            .map(|(_, consolidation)| consolidation.as_inner().clone())
+            .collect()
+    }
 }
 
 /// Filter up to a maximum number of operations out of an iterator.

beacon_node/operation_pool/src/persistence.rs

@@ -193,6 +193,8 @@ impl<T: EthSpec> PersistedOperationPool<T> {
             proposer_slashings,
             voluntary_exits,
             bls_to_execution_changes: RwLock::new(bls_to_execution_changes),
+            // TODO(maxeb): persist consolidations
+            consolidations: <_>::default(),
             reward_cache: Default::default(),
             _phantom: Default::default(),
         };

consensus/state_processing/src/per_block_processing.rs

@@ -19,6 +19,7 @@ pub use verify_attestation::{
     verify_attestation_for_block_inclusion, verify_attestation_for_state,
 };
 pub use verify_bls_to_execution_change::verify_bls_to_execution_change;
+pub use verify_consolidation::verify_consolidation;
 pub use verify_deposit::{
     get_existing_validator_index, verify_deposit_merkle_proof, verify_deposit_signature,
 };

consensus/state_processing/src/per_block_processing/errors.rs

@@ -207,6 +207,7 @@ pub type SyncCommitteeMessageValidationError = BlockOperationError<SyncAggregate
 pub type DepositValidationError = BlockOperationError<DepositInvalid>;
 pub type ExitValidationError = BlockOperationError<ExitInvalid>;
 pub type BlsExecutionChangeValidationError = BlockOperationError<BlsExecutionChangeInvalid>;
+pub type ConsolidationValidationError = BlockOperationError<ConsolidationInvalid>;
 
 #[derive(Debug, PartialEq, Clone)]
 pub enum BlockOperationError<T> {

consensus/state_processing/src/per_block_processing/verify_consolidation.rs

@@ -6,7 +6,7 @@ use crate::{
 use types::*;
 
 pub fn verify_consolidation<T: EthSpec>(
-    state: &mut BeaconState<T>,
+    state: &BeaconState<T>,
     signed_consolidation: &SignedConsolidation,
     verify_signatures: VerifySignatures,
     spec: &ChainSpec,

consensus/state_processing/src/verify_operation.rs

@@ -1,9 +1,9 @@
 use crate::per_block_processing::{
     errors::{
-        AttesterSlashingValidationError, BlsExecutionChangeValidationError, ExitValidationError,
-        ProposerSlashingValidationError,
+        AttesterSlashingValidationError, BlsExecutionChangeValidationError,
+        ConsolidationValidationError, ExitValidationError, ProposerSlashingValidationError,
     },
-    verify_attester_slashing, verify_bls_to_execution_change, verify_exit,
+    verify_attester_slashing, verify_bls_to_execution_change, verify_consolidation, verify_exit,
     verify_proposer_slashing,
 };
 use crate::VerifySignatures;
@@ -14,7 +14,7 @@ use ssz_derive::{Decode, Encode};
 use std::marker::PhantomData;
 use types::{
     AttesterSlashing, BeaconState, ChainSpec, Epoch, EthSpec, Fork, ForkVersion, ProposerSlashing,
-    SignedBlsToExecutionChange, SignedVoluntaryExit,
+    SignedBlsToExecutionChange, SignedConsolidation, SignedVoluntaryExit,
 };
 
 const MAX_FORKS_VERIFIED_AGAINST: usize = 2;
@@ -206,6 +206,24 @@ impl<E: EthSpec> VerifyOperation<E> for SignedBlsToExecutionChange {
     }
 }
 
+impl<E: EthSpec> VerifyOperation<E> for SignedConsolidation {
+    type Error = ConsolidationValidationError;
+
+    fn validate(
+        self,
+        state: &BeaconState<E>,
+        spec: &ChainSpec,
+    ) -> Result<SigVerifiedOp<Self, E>, Self::Error> {
+        verify_consolidation(state, &self, VerifySignatures::True, spec)?;
+        Ok(SigVerifiedOp::new(self, state))
+    }
+
+    #[allow(clippy::arithmetic_side_effects)]
+    fn verification_epochs(&self) -> SmallVec<[Epoch; MAX_FORKS_VERIFIED_AGAINST]> {
+        smallvec![self.message.epoch]
+    }
+}
+
 /// Trait for operations that can be verified and transformed into a
 /// `SigVerifiedOp`.
 ///