Rework block processing (sigp#4092)

* introduce availability pending block

* add intoavailableblock trait

* small fixes

* add 'gossip blob cache' and start to clean up processing and transition types

* shard memory blob cache

* Initial commit

* Fix after rebase

* Add gossip verification conditions

* cache cleanup

* general chaos

* extended chaos

* cargo fmt

* more progress

* more progress

* tons of changes, just tryna compile

* everything, everywhere, all at once

* Reprocess an ExecutedBlock on unavailable blobs

* Add sus gossip verification for blobs

* Merge stuff

* Remove reprocessing cache stuff

* lint

* Add a wrapper to allow construction of only valid `AvailableBlock`s

* rename blob arc list to blob list

* merge cleanuo

* Revert "merge cleanuo"

This reverts commit 5e98326.

* Revert "Revert "merge cleanuo""

This reverts commit 3a40094.

* fix rpc methods

* move beacon block and blob to eth2/types

* rename gossip blob cache to data availability checker

* lots of changes

* fix some compilation issues

* fix compilation issues

* fix compilation issues

* fix compilation issues

* fix compilation issues

* fix compilation issues

* cargo fmt

* use a common data structure for block import types

* fix availability check on proposal import

* refactor the blob cache and split the block wrapper into two types

* add type conversion for signed block and block wrapper

* fix beacon chain tests and do some renaming, add some comments

* Partial processing (#4)

* move beacon block and blob to eth2/types

* rename gossip blob cache to data availability checker

* lots of changes

* fix some compilation issues

* fix compilation issues

* fix compilation issues

* fix compilation issues

* fix compilation issues

* fix compilation issues

* cargo fmt

* use a common data structure for block import types

* fix availability check on proposal import

* refactor the blob cache and split the block wrapper into two types

* add type conversion for signed block and block wrapper

* fix beacon chain tests and do some renaming, add some comments

* cargo update (#6)

---------

Co-authored-by: realbigsean <sean@sigmaprime.io>
Co-authored-by: realbigsean <seananderson33@gmail.com>

beacon_node/beacon_chain/src/block_verification.rs

@@ -24,9 +24,6 @@
 //!             ▼
 //!     SignedBeaconBlock
 //!             |
-//!             ▼
-//!      AvailableBlock
-//!             |
 //!             |---------------
 //!             |              |
 //!             |              ▼
@@ -51,9 +48,9 @@
 // returned alongside.
 #![allow(clippy::result_large_err)]
 
-use crate::blob_verification::{
-    validate_blob_for_gossip, AsBlock, AvailableBlock, BlobError, BlockWrapper, IntoAvailableBlock,
-    IntoBlockWrapper,
+use crate::blob_verification::{AsBlock, BlobError, BlockWrapper, MaybeAvailableBlock};
+use crate::data_availability_checker::{
+    AvailabilityCheckError, AvailabilityPendingBlock, AvailableBlock,
 };
 use crate::eth1_finalization_cache::Eth1FinalizationData;
 use crate::execution_payload::{
@@ -96,7 +93,6 @@ use std::time::Duration;
 use store::{Error as DBError, HotStateSummary, KeyValueStore, StoreOp};
 use task_executor::JoinHandle;
 use tree_hash::TreeHash;
-use types::signed_beacon_block::BlobReconstructionError;
 use types::ExecPayload;
 use types::{
     BeaconBlockRef, BeaconState, BeaconStateError, BlindedPayload, ChainSpec, CloneConfig, Epoch,
@@ -314,6 +310,7 @@ pub enum BlockError<T: EthSpec> {
         parent_root: Hash256,
     },
     BlobValidation(BlobError),
+    AvailabilityCheck(AvailabilityCheckError),
 }
 
 impl<T: EthSpec> From<BlobError> for BlockError<T> {
@@ -322,6 +319,12 @@ impl<T: EthSpec> From<BlobError> for BlockError<T> {
     }
 }
 
+impl<T: EthSpec> From<AvailabilityCheckError> for BlockError<T> {
+    fn from(e: AvailabilityCheckError) -> Self {
+        Self::AvailabilityCheck(e)
+    }
+}
+
 /// Returned when block validation failed due to some issue verifying
 /// the execution payload.
 #[derive(Debug)]
@@ -494,13 +497,8 @@ impl<T: EthSpec> From<ArithError> for BlockError<T> {
     }
 }
 
-impl<T: EthSpec> From<BlobReconstructionError> for BlockError<T> {
-    fn from(e: BlobReconstructionError) -> Self {
-        BlockError::BlobValidation(BlobError::from(e))
-    }
-}
-
 /// Stores information about verifying a payload against an execution engine.
+#[derive(Clone)]
 pub struct PayloadVerificationOutcome {
     pub payload_verification_status: PayloadVerificationStatus,
     pub is_valid_merge_transition_block: bool,
@@ -605,14 +603,14 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
 
         signature_verifier.include_all_signatures(block.as_block(), &mut consensus_context)?;
 
-        //FIXME(sean) batch kzg verification
-        let available_block = block.clone().into_available_block(*block_root, chain)?;
-        consensus_context = consensus_context.set_kzg_commitments_consistent(true);
+        let maybe_available_block = chain
+            .data_availability_checker
+            .check_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.
         signature_verified_blocks.push(SignatureVerifiedBlock {
-            block: available_block,
+            block: maybe_available_block,
             block_root: *block_root,
             parent: None,
             consensus_context,
@@ -637,7 +635,7 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
 #[derive(Derivative)]
 #[derivative(Debug(bound = "T: BeaconChainTypes"))]
 pub struct GossipVerifiedBlock<T: BeaconChainTypes> {
-    pub block: AvailableBlock<T::EthSpec>,
+    pub block: MaybeAvailableBlock<T::EthSpec>,
     pub block_root: Hash256,
     parent: Option<PreProcessingSnapshot<T::EthSpec>>,
     consensus_context: ConsensusContext<T::EthSpec>,
@@ -646,7 +644,7 @@ pub struct GossipVerifiedBlock<T: BeaconChainTypes> {
 /// A wrapper around a `SignedBeaconBlock` that indicates that all signatures (except the deposit
 /// signatures) have been verified.
 pub struct SignatureVerifiedBlock<T: BeaconChainTypes> {
-    block: AvailableBlock<T::EthSpec>,
+    block: MaybeAvailableBlock<T::EthSpec>,
     block_root: Hash256,
     parent: Option<PreProcessingSnapshot<T::EthSpec>>,
     consensus_context: ConsensusContext<T::EthSpec>,
@@ -669,14 +667,103 @@ type PayloadVerificationHandle<E> =
 /// due to finality or some other event. A `ExecutionPendingBlock` should be imported into the
 /// `BeaconChain` immediately after it is instantiated.
 pub struct ExecutionPendingBlock<T: BeaconChainTypes> {
-    pub block: AvailableBlock<T::EthSpec>,
+    pub block: MaybeAvailableBlock<T::EthSpec>,
+    pub import_data: BlockImportData<T::EthSpec>,
+    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,
+                ))
+            }
+        }
+    }
+}
+
+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 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 struct BlockImportData<E: EthSpec> {
     pub block_root: Hash256,
-    pub state: BeaconState<T::EthSpec>,
-    pub parent_block: SignedBeaconBlock<T::EthSpec, BlindedPayload<T::EthSpec>>,
+    pub state: BeaconState<E>,
+    pub parent_block: SignedBeaconBlock<E, BlindedPayload<E>>,
     pub parent_eth1_finalization_data: Eth1FinalizationData,
     pub confirmed_state_roots: Vec<Hash256>,
-    pub consensus_context: ConsensusContext<T::EthSpec>,
-    pub payload_verification_handle: PayloadVerificationHandle<T::EthSpec>,
+    pub consensus_context: ConsensusContext<E>,
 }
 
 /// Implemented on types that can be converted into a `ExecutionPendingBlock`.
@@ -720,19 +807,20 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
         block: BlockWrapper<T::EthSpec>,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockError<T::EthSpec>> {
+        let maybe_available = chain.data_availability_checker.check_availability(block)?;
         // 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 = block.signed_block_header();
-        Self::new_without_slasher_checks(block, chain).map_err(|e| {
+        let header = maybe_available.signed_block_header();
+        Self::new_without_slasher_checks(maybe_available, 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: BlockWrapper<T::EthSpec>,
+        block: MaybeAvailableBlock<T::EthSpec>,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockError<T::EthSpec>> {
         // Ensure the block is the correct structure for the fork at `block.slot()`.
@@ -929,16 +1017,14 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
         // Validate the block's execution_payload (if any).
         validate_execution_payload_for_gossip(&parent_block, block.message(), chain)?;
 
-        let available_block = validate_blob_for_gossip(block, block_root, chain)?;
-
         // Having checked the proposer index and the block root we can cache them.
-        let consensus_context = ConsensusContext::new(available_block.slot())
+        let consensus_context = ConsensusContext::new(block.slot())
             .set_current_block_root(block_root)
-            .set_proposer_index(available_block.as_block().message().proposer_index())
+            .set_proposer_index(block.as_block().message().proposer_index())
             .set_kzg_commitments_consistent(true);
 
         Ok(Self {
-            block: available_block,
+            block,
             block_root,
             parent,
             consensus_context,
@@ -978,10 +1064,11 @@ 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: AvailableBlock<T::EthSpec>,
+        block: BlockWrapper<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()
@@ -1028,7 +1115,7 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
 
     /// As for `new` above but producing `BlockSlashInfo`.
     pub fn check_slashable(
-        block: AvailableBlock<T::EthSpec>,
+        block: BlockWrapper<T::EthSpec>,
         block_root: Hash256,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockSlashInfo<BlockError<T::EthSpec>>> {
@@ -1137,12 +1224,7 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for Arc<SignedBeaconBlock
         let block_root = check_block_relevancy(&self, block_root, chain)
             .map_err(|e| BlockSlashInfo::SignatureNotChecked(self.signed_block_header(), e))?;
 
-        let header = self.signed_block_header();
-        let available_block = BlockWrapper::from(self)
-            .into_available_block(block_root, chain)
-            .map_err(|e| BlockSlashInfo::from_early_error(header, BlockError::BlobValidation(e)))?;
-
-        SignatureVerifiedBlock::check_slashable(available_block, block_root, chain)?
+        SignatureVerifiedBlock::check_slashable(self.into(), block_root, chain)?
             .into_execution_pending_block_slashable(block_root, chain, notify_execution_layer)
     }
 
@@ -1151,7 +1233,7 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for Arc<SignedBeaconBlock
     }
 }
 
-impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for AvailableBlock<T::EthSpec> {
+impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for BlockWrapper<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(
@@ -1182,7 +1264,7 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
     ///
     /// Returns an error if the block is invalid, or if the block was unable to be verified.
     pub fn from_signature_verified_components(
-        block: AvailableBlock<T::EthSpec>,
+        block: MaybeAvailableBlock<T::EthSpec>,
         block_root: Hash256,
         parent: PreProcessingSnapshot<T::EthSpec>,
         mut consensus_context: ConsensusContext<T::EthSpec>,
@@ -1562,12 +1644,14 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
 
         Ok(Self {
             block,
-            block_root,
-            state,
-            parent_block: parent.beacon_block,
-            parent_eth1_finalization_data,
-            confirmed_state_roots,
-            consensus_context,
+            import_data: BlockImportData {
+                block_root,
+                state,
+                parent_block: parent.beacon_block,
+                parent_eth1_finalization_data,
+                confirmed_state_roots,
+                consensus_context,
+            },
             payload_verification_handle,
         })
     }
@@ -1648,7 +1732,7 @@ fn check_block_against_finalized_slot<T: BeaconChainTypes>(
 /// Taking a lock on the `chain.canonical_head.fork_choice` might cause a deadlock here.
 pub fn check_block_is_finalized_checkpoint_or_descendant<
     T: BeaconChainTypes,
-    B: IntoBlockWrapper<T::EthSpec>,
+    B: AsBlock<T::EthSpec>,
 >(
     chain: &BeaconChain<T>,
     fork_choice: &BeaconForkChoice<T>,
@@ -1746,16 +1830,16 @@ 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: BlockWrapper<T::EthSpec>,
-) -> Result<(ProtoBlock, BlockWrapper<T::EthSpec>), BlockError<T::EthSpec>> {
+    block: MaybeAvailableBlock<T::EthSpec>,
+) -> Result<(ProtoBlock, MaybeAvailableBlock<T::EthSpec>), BlockError<T::EthSpec>> {
     if let Some(proto_block) = chain
         .canonical_head
         .fork_choice_read_lock()
         .get_block(&block.parent_root())
     {
         Ok((proto_block, block))
     } else {
-        Err(BlockError::ParentUnknown(block))
+        Err(BlockError::ParentUnknown(block.into_block_wrapper()))
     }
 }
 
@@ -1764,7 +1848,7 @@ fn verify_parent_block_is_known<T: BeaconChainTypes>(
 /// Returns `Err(BlockError::ParentUnknown)` if the parent is not found, or if an error occurs
 /// whilst attempting the operation.
 #[allow(clippy::type_complexity)]
-fn load_parent<T: BeaconChainTypes, B: IntoBlockWrapper<T::EthSpec>>(
+fn load_parent<T: BeaconChainTypes, B: AsBlock<T::EthSpec>>(
     block_root: Hash256,
     block: B,
     chain: &BeaconChain<T>,