finish refactor

beacon_node/network/src/beacon_processor/mod.rs

@@ -40,6 +40,7 @@
 
 use crate::sync::manager::BlockProcessType;
 use crate::{metrics, service::NetworkMessage, sync::SyncMessage};
+use beacon_chain::blob_verification::BlockWrapper;
 use beacon_chain::parking_lot::Mutex;
 use beacon_chain::{BeaconChain, BeaconChainTypes, GossipVerifiedBlock, NotifyExecutionLayer};
 use derivative::Derivative;
@@ -62,7 +63,6 @@ use std::time::Duration;
 use std::{cmp, collections::HashSet};
 use task_executor::TaskExecutor;
 use tokio::sync::mpsc;
-use types::signed_block_and_blobs::BlockWrapper;
 use types::{
     Attestation, AttesterSlashing, Hash256, LightClientFinalityUpdate, LightClientOptimisticUpdate,
     ProposerSlashing, SignedAggregateAndProof, SignedBeaconBlock, SignedBeaconBlockAndBlobsSidecar,

beacon_node/network/src/beacon_processor/work_reprocessing_queue.rs

@@ -13,6 +13,7 @@
 use super::MAX_SCHEDULED_WORK_QUEUE_LEN;
 use crate::metrics;
 use crate::sync::manager::BlockProcessType;
+use beacon_chain::blob_verification::{AsBlock, BlockWrapper};
 use beacon_chain::{BeaconChainTypes, GossipVerifiedBlock, MAXIMUM_GOSSIP_CLOCK_DISPARITY};
 use fnv::FnvHashMap;
 use futures::task::Poll;
@@ -29,7 +30,6 @@ use task_executor::TaskExecutor;
 use tokio::sync::mpsc::{self, Receiver, Sender};
 use tokio::time::error::Error as TimeError;
 use tokio_util::time::delay_queue::{DelayQueue, Key as DelayKey};
-use types::signed_block_and_blobs::BlockWrapper;
 use types::{Attestation, EthSpec, Hash256, SignedAggregateAndProof, SubnetId};
 
 const TASK_NAME: &str = "beacon_processor_reprocess_queue";

beacon_node/network/src/beacon_processor/worker/gossip_methods.rs

@@ -1,5 +1,6 @@
 use crate::{metrics, service::NetworkMessage, sync::SyncMessage};
 
+use beacon_chain::blob_verification::{AsBlock, BlockWrapper};
 use beacon_chain::store::Error;
 use beacon_chain::{
     attestation_verification::{self, Error as AttnError, VerifiedAttestation},
@@ -18,7 +19,6 @@ use ssz::Encode;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use store::hot_cold_store::HotColdDBError;
 use tokio::sync::mpsc;
-use types::signed_block_and_blobs::BlockWrapper;
 use types::{
     Attestation, AttesterSlashing, EthSpec, Hash256, IndexedAttestation, LightClientFinalityUpdate,
     LightClientOptimisticUpdate, ProposerSlashing, SignedAggregateAndProof,
@@ -726,7 +726,7 @@ impl<T: BeaconChainTypes> Worker<T> {
         let block_root = if let Ok(verified_block) = &verification_result {
             verified_block.block_root
         } else {
-            block.block().canonical_root()
+            block.as_block().canonical_root()
         };
 
         // Write the time the block was observed into delay cache.

beacon_node/network/src/beacon_processor/worker/sync_methods.rs

@@ -7,6 +7,7 @@ use crate::beacon_processor::DuplicateCache;
 use crate::metrics;
 use crate::sync::manager::{BlockProcessType, SyncMessage};
 use crate::sync::{BatchProcessResult, ChainId};
+use beacon_chain::blob_verification::{AsBlock, BlockWrapper, IntoAvailableBlock};
 use beacon_chain::CountUnrealized;
 use beacon_chain::{
     BeaconChainError, BeaconChainTypes, BlockError, ChainSegmentResult, HistoricalBlockError,
@@ -16,7 +17,6 @@ use lighthouse_network::PeerAction;
 use slog::{debug, error, info, warn};
 use std::sync::Arc;
 use tokio::sync::mpsc;
-use types::signed_block_and_blobs::BlockWrapper;
 use types::{Epoch, Hash256, SignedBeaconBlock};
 
 /// Id associated to a batch processing request, either a sync batch or a parent lookup.
@@ -85,15 +85,23 @@ impl<T: BeaconChainTypes> Worker<T> {
             }
         };
         let slot = block.slot();
-        let result = self
-            .chain
-            .process_block(
-                block_root,
-                block,
-                CountUnrealized::True,
-                NotifyExecutionLayer::Yes,
-            )
-            .await;
+        let available_block = block
+            .into_available_block(block_root, &self.chain)
+            .map_err(BlockError::BlobValidation);
+
+        let result = match available_block {
+            Ok(block) => {
+                self.chain
+                    .process_block(
+                        block_root,
+                        block,
+                        CountUnrealized::True,
+                        NotifyExecutionLayer::Yes,
+                    )
+                    .await
+            }
+            Err(e) => Err(e),
+        };
 
         metrics::inc_counter(&metrics::BEACON_PROCESSOR_RPC_BLOCK_IMPORTED_TOTAL);