Refactor deneb block processing (#4511)

* Revert "fix merge"

This reverts commit 405e95b0ce.

* refactor deneb block processing

* cargo fmt

* fix ci

beacon_node/beacon_chain/src/block_verification.rs

@@ -48,13 +48,11 @@
 // returned alongside.
 #![allow(clippy::result_large_err)]
 
-use crate::blob_verification::{
-    AsBlock, BlobError, BlockWrapper, GossipVerifiedBlob, GossipVerifiedBlobList,
-    MaybeAvailableBlock,
-};
-use crate::data_availability_checker::{
-    AvailabilityCheckError, AvailabilityPendingBlock, AvailableBlock,
+use crate::blob_verification::{BlobError, GossipVerifiedBlob};
+use crate::block_verification_types::{
+    AsBlock, BlockImportData, GossipVerifiedBlockContents, RpcBlock,
 };
+use crate::data_availability_checker::{AvailabilityCheckError, MaybeAvailableBlock};
 use crate::eth1_finalization_cache::Eth1FinalizationData;
 use crate::execution_payload::{
     is_optimistic_candidate_block, validate_execution_payload_for_gossip, validate_merge_block,
@@ -99,13 +97,11 @@ use std::time::Duration;
 use store::{Error as DBError, HotStateSummary, KeyValueStore, SignedBlobSidecarList, StoreOp};
 use task_executor::JoinHandle;
 use tree_hash::TreeHash;
-use types::blob_sidecar::BlobIdentifier;
 use types::ExecPayload;
-use types::{ssz_tagged_beacon_state, ssz_tagged_signed_beacon_block};
 use types::{
-    BeaconBlockRef, BeaconState, BeaconStateError, BlindedPayload, ChainSpec, CloneConfig, Epoch,
-    EthSpec, ExecutionBlockHash, Hash256, InconsistentFork, PublicKey, PublicKeyBytes,
-    RelativeEpoch, SignedBeaconBlock, SignedBeaconBlockHeader, Slot,
+    BeaconBlockRef, BeaconState, BeaconStateError, ChainSpec, CloneConfig, Epoch, EthSpec,
+    ExecutionBlockHash, Hash256, InconsistentFork, PublicKey, PublicKeyBytes, RelativeEpoch,
+    SignedBeaconBlock, SignedBeaconBlockHeader, Slot,
 };
 
 pub const POS_PANDA_BANNER: &str = r#"
@@ -153,7 +149,7 @@ pub enum BlockError<T: EthSpec> {
     ///
     /// It's unclear if this block is valid, but it cannot be processed without already knowing
     /// its parent.
-    ParentUnknown(BlockWrapper<T>),
+    ParentUnknown(RpcBlock<T>),
     /// The block slot is greater than the present slot.
     ///
     /// ## Peer scoring
@@ -558,7 +554,7 @@ fn process_block_slash_info<T: BeaconChainTypes>(
 /// The given `chain_segment` must contain only blocks from the same epoch, otherwise an error
 /// will be returned.
 pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
-    mut chain_segment: Vec<(Hash256, BlockWrapper<T::EthSpec>)>,
+    mut chain_segment: Vec<(Hash256, RpcBlock<T::EthSpec>)>,
     chain: &BeaconChain<T>,
 ) -> Result<Vec<SignatureVerifiedBlock<T>>, BlockError<T::EthSpec>> {
     if chain_segment.is_empty() {
@@ -595,7 +591,7 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
 
         let maybe_available_block = chain
             .data_availability_checker
-            .check_availability(block.clone())?;
+            .check_rpc_block_availability(block.clone())?;
 
         // Save the block and its consensus context. The context will have had its proposer index
         // and attesting indices filled in, which can be used to accelerate later block processing.
@@ -625,19 +621,12 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
 #[derive(Derivative)]
 #[derivative(Debug(bound = "T: BeaconChainTypes"))]
 pub struct GossipVerifiedBlock<T: BeaconChainTypes> {
-    pub block: MaybeAvailableBlock<T::EthSpec>,
+    pub block: Arc<SignedBeaconBlock<T::EthSpec>>,
     pub block_root: Hash256,
     parent: Option<PreProcessingSnapshot<T::EthSpec>>,
     consensus_context: ConsensusContext<T::EthSpec>,
 }
 
-impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
-    /// Useful for publishing after gossip verification.
-    pub fn into_block_wrapper(self) -> BlockWrapper<T::EthSpec> {
-        self.block.into_block_wrapper()
-    }
-}
-
 /// A wrapper around a `SignedBeaconBlock` that indicates that all signatures (except the deposit
 /// signatures) have been verified.
 pub struct SignatureVerifiedBlock<T: BeaconChainTypes> {
@@ -669,147 +658,6 @@ pub struct ExecutionPendingBlock<T: BeaconChainTypes> {
     pub payload_verification_handle: PayloadVerificationHandle<T::EthSpec>,
 }
 
-pub enum ExecutedBlock<E: EthSpec> {
-    Available(AvailableExecutedBlock<E>),
-    AvailabilityPending(AvailabilityPendingExecutedBlock<E>),
-}
-
-impl<E: EthSpec> ExecutedBlock<E> {
-    pub fn as_block(&self) -> &SignedBeaconBlock<E> {
-        match self {
-            Self::Available(available) => available.block.block(),
-            Self::AvailabilityPending(pending) => pending.block.as_block(),
-        }
-    }
-}
-
-impl<E: EthSpec> std::fmt::Debug for ExecutedBlock<E> {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{:?}", self.as_block())
-    }
-}
-
-impl<E: EthSpec> ExecutedBlock<E> {
-    pub fn new(
-        block: MaybeAvailableBlock<E>,
-        import_data: BlockImportData<E>,
-        payload_verification_outcome: PayloadVerificationOutcome,
-    ) -> Self {
-        match block {
-            MaybeAvailableBlock::Available(available_block) => {
-                Self::Available(AvailableExecutedBlock::new(
-                    available_block,
-                    import_data,
-                    payload_verification_outcome,
-                ))
-            }
-            MaybeAvailableBlock::AvailabilityPending(pending_block) => {
-                Self::AvailabilityPending(AvailabilityPendingExecutedBlock::new(
-                    pending_block,
-                    import_data,
-                    payload_verification_outcome,
-                ))
-            }
-        }
-    }
-}
-
-#[derive(Debug, PartialEq)]
-pub struct AvailableExecutedBlock<E: EthSpec> {
-    pub block: AvailableBlock<E>,
-    pub import_data: BlockImportData<E>,
-    pub payload_verification_outcome: PayloadVerificationOutcome,
-}
-
-impl<E: EthSpec> AvailableExecutedBlock<E> {
-    pub fn new(
-        block: AvailableBlock<E>,
-        import_data: BlockImportData<E>,
-        payload_verification_outcome: PayloadVerificationOutcome,
-    ) -> Self {
-        Self {
-            block,
-            import_data,
-            payload_verification_outcome,
-        }
-    }
-
-    pub fn get_all_blob_ids(&self) -> Vec<BlobIdentifier> {
-        let num_blobs_expected = self
-            .block
-            .message()
-            .body()
-            .blob_kzg_commitments()
-            .map_or(0, |commitments| commitments.len());
-        let mut blob_ids = Vec::with_capacity(num_blobs_expected);
-        for i in 0..num_blobs_expected {
-            blob_ids.push(BlobIdentifier {
-                block_root: self.import_data.block_root,
-                index: i as u64,
-            });
-        }
-        blob_ids
-    }
-}
-
-#[derive(Encode, Decode, Clone)]
-pub struct AvailabilityPendingExecutedBlock<E: EthSpec> {
-    pub block: AvailabilityPendingBlock<E>,
-    pub import_data: BlockImportData<E>,
-    pub payload_verification_outcome: PayloadVerificationOutcome,
-}
-
-impl<E: EthSpec> AvailabilityPendingExecutedBlock<E> {
-    pub fn new(
-        block: AvailabilityPendingBlock<E>,
-        import_data: BlockImportData<E>,
-        payload_verification_outcome: PayloadVerificationOutcome,
-    ) -> Self {
-        Self {
-            block,
-            import_data,
-            payload_verification_outcome,
-        }
-    }
-
-    pub fn num_blobs_expected(&self) -> usize {
-        self.block
-            .kzg_commitments()
-            .map_or(0, |commitments| commitments.len())
-    }
-
-    pub fn get_all_blob_ids(&self) -> Vec<BlobIdentifier> {
-        let block_root = self.import_data.block_root;
-        self.block
-            .get_filtered_blob_ids(Some(block_root), |_, _| true)
-    }
-
-    pub fn get_filtered_blob_ids(
-        &self,
-        filter: impl Fn(usize, Hash256) -> bool,
-    ) -> Vec<BlobIdentifier> {
-        self.block
-            .get_filtered_blob_ids(Some(self.import_data.block_root), filter)
-    }
-}
-
-#[derive(Debug, PartialEq, Encode, Decode, Clone)]
-// TODO (mark): investigate using an Arc<state> / Arc<parent_block>
-//              here to make this cheaper to clone
-pub struct BlockImportData<E: EthSpec> {
-    pub block_root: Hash256,
-    #[ssz(with = "ssz_tagged_beacon_state")]
-    pub state: BeaconState<E>,
-    #[ssz(with = "ssz_tagged_signed_beacon_block")]
-    pub parent_block: SignedBeaconBlock<E, BlindedPayload<E>>,
-    pub parent_eth1_finalization_data: Eth1FinalizationData,
-    pub confirmed_state_roots: Vec<Hash256>,
-    pub consensus_context: ConsensusContext<E>,
-}
-
-pub type GossipVerifiedBlockContents<T> =
-    (GossipVerifiedBlock<T>, Option<GossipVerifiedBlobList<T>>);
-
 pub trait IntoGossipVerifiedBlockContents<T: BeaconChainTypes>: Sized {
     fn into_gossip_verified_block(
         self,
@@ -911,27 +759,23 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
         block: Arc<SignedBeaconBlock<T::EthSpec>>,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockError<T::EthSpec>> {
-        let maybe_available = chain
-            .data_availability_checker
-            .check_availability(block.into())?;
         // If the block is valid for gossip we don't supply it to the slasher here because
         // we assume it will be transformed into a fully verified block. We *do* need to supply
         // it to the slasher if an error occurs, because that's the end of this block's journey,
         // and it could be a repeat proposal (a likely cause for slashing!).
-        let header = maybe_available.signed_block_header();
-        Self::new_without_slasher_checks(maybe_available, chain).map_err(|e| {
+        let header = block.signed_block_header();
+        Self::new_without_slasher_checks(block, chain).map_err(|e| {
             process_block_slash_info(chain, BlockSlashInfo::from_early_error(header, e))
         })
     }
 
     /// As for new, but doesn't pass the block to the slasher.
     fn new_without_slasher_checks(
-        block: MaybeAvailableBlock<T::EthSpec>,
+        block: Arc<SignedBeaconBlock<T::EthSpec>>,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockError<T::EthSpec>> {
         // Ensure the block is the correct structure for the fork at `block.slot()`.
         block
-            .as_block()
             .fork_name(&chain.spec)
             .map_err(BlockError::InconsistentFork)?;
 
@@ -947,7 +791,7 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
             });
         }
 
-        let block_root = get_block_root(block.as_block());
+        let block_root = get_block_root(&block);
 
         // Disallow blocks that conflict with the anchor (weak subjectivity checkpoint), if any.
         check_block_against_anchor_slot(block.message(), chain)?;
@@ -1067,7 +911,7 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
             let pubkey = pubkey_cache
                 .get(block.message().proposer_index() as usize)
                 .ok_or_else(|| BlockError::UnknownValidator(block.message().proposer_index()))?;
-            block.as_block().verify_signature(
+            block.verify_signature(
                 Some(block_root),
                 pubkey,
                 &fork,
@@ -1111,8 +955,7 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
         // Having checked the proposer index and the block root we can cache them.
         let consensus_context = ConsensusContext::new(block.slot())
             .set_current_block_root(block_root)
-            .set_proposer_index(block.as_block().message().proposer_index())
-            .set_kzg_commitments_consistent(true);
+            .set_proposer_index(block.as_block().message().proposer_index());
 
         Ok(Self {
             block,
@@ -1155,11 +998,10 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
     ///
     /// Returns an error if the block is invalid, or if the block was unable to be verified.
     pub fn new(
-        block: BlockWrapper<T::EthSpec>,
+        block: MaybeAvailableBlock<T::EthSpec>,
         block_root: Hash256,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockError<T::EthSpec>> {
-        let block = chain.data_availability_checker.check_availability(block)?;
         // Ensure the block is the correct structure for the fork at `block.slot()`.
         block
             .as_block()
@@ -1182,10 +1024,8 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
 
         let mut signature_verifier = get_signature_verifier(&state, &pubkey_cache, &chain.spec);
 
-        let mut consensus_context = ConsensusContext::new(block.slot())
-            .set_current_block_root(block_root)
-            // An `AvailabileBlock is passed in here, so we know this check has been run.`
-            .set_kzg_commitments_consistent(true);
+        let mut consensus_context =
+            ConsensusContext::new(block.slot()).set_current_block_root(block_root);
 
         signature_verifier.include_all_signatures(block.as_block(), &mut consensus_context)?;
 
@@ -1203,7 +1043,7 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
 
     /// As for `new` above but producing `BlockSlashInfo`.
     pub fn check_slashable(
-        block: BlockWrapper<T::EthSpec>,
+        block: MaybeAvailableBlock<T::EthSpec>,
         block_root: Hash256,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockSlashInfo<BlockError<T::EthSpec>>> {
@@ -1238,11 +1078,11 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
         // signature.
         let mut consensus_context = from.consensus_context;
         signature_verifier
-            .include_all_signatures_except_proposal(block.as_block(), &mut consensus_context)?;
+            .include_all_signatures_except_proposal(block.as_ref(), &mut consensus_context)?;
 
         if signature_verifier.verify().is_ok() {
             Ok(Self {
-                block,
+                block: MaybeAvailableBlock::AvailabilityPending(block),
                 block_root: from.block_root,
                 parent: Some(parent),
                 consensus_context,
@@ -1299,6 +1139,7 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for SignatureVerifiedBloc
     }
 }
 
+//TODO(sean) can this be deleted
 impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for Arc<SignedBeaconBlock<T::EthSpec>> {
     /// Verifies the `SignedBeaconBlock` by first transforming it into a `SignatureVerifiedBlock`
     /// and then using that implementation of `IntoExecutionPendingBlock` to complete verification.
@@ -1311,8 +1152,16 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for Arc<SignedBeaconBlock
         // Perform an early check to prevent wasting time on irrelevant blocks.
         let block_root = check_block_relevancy(&self, block_root, chain)
             .map_err(|e| BlockSlashInfo::SignatureNotChecked(self.signed_block_header(), e))?;
-
-        SignatureVerifiedBlock::check_slashable(self.into(), block_root, chain)?
+        let maybe_available = chain
+            .data_availability_checker
+            .check_rpc_block_availability(RpcBlock::new_without_blobs(self.clone()))
+            .map_err(|e| {
+                BlockSlashInfo::SignatureNotChecked(
+                    self.signed_block_header(),
+                    BlockError::AvailabilityCheck(e),
+                )
+            })?;
+        SignatureVerifiedBlock::check_slashable(maybe_available, block_root, chain)?
             .into_execution_pending_block_slashable(block_root, chain, notify_execution_layer)
     }
 
@@ -1321,7 +1170,7 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for Arc<SignedBeaconBlock
     }
 }
 
-impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for BlockWrapper<T::EthSpec> {
+impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for RpcBlock<T::EthSpec> {
     /// Verifies the `SignedBeaconBlock` by first transforming it into a `SignatureVerifiedBlock`
     /// and then using that implementation of `IntoExecutionPendingBlock` to complete verification.
     fn into_execution_pending_block_slashable(
@@ -1333,8 +1182,16 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for BlockWrapper<T::EthSp
         // Perform an early check to prevent wasting time on irrelevant blocks.
         let block_root = check_block_relevancy(self.as_block(), block_root, chain)
             .map_err(|e| BlockSlashInfo::SignatureNotChecked(self.signed_block_header(), e))?;
-
-        SignatureVerifiedBlock::check_slashable(self, block_root, chain)?
+        let maybe_available = chain
+            .data_availability_checker
+            .check_rpc_block_availability(self.clone())
+            .map_err(|e| {
+                BlockSlashInfo::SignatureNotChecked(
+                    self.signed_block_header(),
+                    BlockError::AvailabilityCheck(e),
+                )
+            })?;
+        SignatureVerifiedBlock::check_slashable(maybe_available, block_root, chain)?
             .into_execution_pending_block_slashable(block_root, chain, notify_execution_layer)
     }
 
@@ -1388,7 +1245,7 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
             //  because it will revert finalization. Note that the finalized block is stored in fork
             //  choice, so we will not reject any child of the finalized block (this is relevant during
             //  genesis).
-            return Err(BlockError::ParentUnknown(block.into_block_wrapper()));
+            return Err(BlockError::ParentUnknown(block.into_rpc_block()));
         }
 
         /*
@@ -1826,7 +1683,7 @@ pub fn check_block_is_finalized_checkpoint_or_descendant<
                 block_parent_root: block.parent_root(),
             })
         } else {
-            Err(BlockError::ParentUnknown(block.into_block_wrapper()))
+            Err(BlockError::ParentUnknown(block.into_rpc_block()))
         }
     }
 }
@@ -1898,8 +1755,8 @@ pub fn get_block_root<E: EthSpec>(block: &SignedBeaconBlock<E>) -> Hash256 {
 #[allow(clippy::type_complexity)]
 fn verify_parent_block_is_known<T: BeaconChainTypes>(
     chain: &BeaconChain<T>,
-    block: MaybeAvailableBlock<T::EthSpec>,
-) -> Result<(ProtoBlock, MaybeAvailableBlock<T::EthSpec>), BlockError<T::EthSpec>> {
+    block: Arc<SignedBeaconBlock<T::EthSpec>>,
+) -> Result<(ProtoBlock, Arc<SignedBeaconBlock<T::EthSpec>>), BlockError<T::EthSpec>> {
     if let Some(proto_block) = chain
         .canonical_head
         .fork_choice_read_lock()
@@ -1907,7 +1764,9 @@ fn verify_parent_block_is_known<T: BeaconChainTypes>(
     {
         Ok((proto_block, block))
     } else {
-        Err(BlockError::ParentUnknown(block.into_block_wrapper()))
+        Err(BlockError::ParentUnknown(RpcBlock::new_without_blobs(
+            block,
+        )))
     }
 }
 
@@ -1938,7 +1797,7 @@ fn load_parent<T: BeaconChainTypes, B: AsBlock<T::EthSpec>>(
         .fork_choice_read_lock()
         .contains_block(&block.parent_root())
     {
-        return Err(BlockError::ParentUnknown(block.into_block_wrapper()));
+        return Err(BlockError::ParentUnknown(block.into_rpc_block()));
     }
 
     let block_delay = chain