add a bunch of blob coupling boiler plate, add a blobs by root request

beacon_node/network/src/beacon_processor/mod.rs

@@ -45,8 +45,7 @@ use beacon_chain::{BeaconChain, BeaconChainTypes, GossipVerifiedBlock};
 use derivative::Derivative;
 use futures::stream::{Stream, StreamExt};
 use futures::task::Poll;
-use lighthouse_network::rpc::methods::BlobsByRangeRequest;
-use lighthouse_network::SignedBeaconBlockAndBlobsSidecar;
+use lighthouse_network::rpc::methods::{BlobsByRangeRequest, BlobsByRootRequest};
 use lighthouse_network::{
     rpc::{BlocksByRangeRequest, BlocksByRootRequest, StatusMessage},
     Client, MessageId, NetworkGlobals, PeerId, PeerRequestId,
@@ -62,11 +61,7 @@ use std::time::Duration;
 use std::{cmp, collections::HashSet};
 use task_executor::TaskExecutor;
 use tokio::sync::mpsc;
-use types::{
-    Attestation, AttesterSlashing, Hash256, ProposerSlashing, SignedAggregateAndProof,
-    SignedBeaconBlock, SignedContributionAndProof, SignedVoluntaryExit, SubnetId,
-    SyncCommitteeMessage, SyncSubnetId,
-};
+use types::{Attestation, AttesterSlashing, Hash256, ProposerSlashing, SignedAggregateAndProof, SignedBeaconBlock, SignedBeaconBlockAndBlobsSidecar, SignedContributionAndProof, SignedVoluntaryExit, SubnetId, SyncCommitteeMessage, SyncSubnetId};
 use work_reprocessing_queue::{
     spawn_reprocess_scheduler, QueuedAggregate, QueuedRpcBlock, QueuedUnaggregate, ReadyWork,
 };
@@ -204,6 +199,7 @@ pub const STATUS_PROCESSING: &str = "status_processing";
 pub const BLOCKS_BY_RANGE_REQUEST: &str = "blocks_by_range_request";
 pub const BLOCKS_BY_ROOTS_REQUEST: &str = "blocks_by_roots_request";
 pub const BLOBS_BY_RANGE_REQUEST: &str = "blobs_by_range_request";
+pub const BLOBS_BY_ROOTS_REQUEST: &str = "blobs_by_roots_request";
 pub const UNKNOWN_BLOCK_ATTESTATION: &str = "unknown_block_attestation";
 pub const UNKNOWN_BLOCK_AGGREGATE: &str = "unknown_block_aggregate";
 
@@ -601,6 +597,21 @@ impl<T: BeaconChainTypes> WorkEvent<T> {
         }
     }
 
+    pub fn blobs_by_root_request(
+        peer_id: PeerId,
+        request_id: PeerRequestId,
+        request: BlobsByRootRequest,
+    ) -> Self {
+        Self {
+            drop_during_sync: false,
+            work: Work::BlobsByRootsRequest {
+                peer_id,
+                request_id,
+                request,
+            },
+        }
+    }
+
     /// Get a `str` representation of the type of work this `WorkEvent` contains.
     pub fn work_type(&self) -> &'static str {
         self.work.str_id()
@@ -789,6 +800,11 @@ pub enum Work<T: BeaconChainTypes> {
         request_id: PeerRequestId,
         request: BlobsByRangeRequest,
     },
+    BlobsByRootsRequest {
+        peer_id: PeerId,
+        request_id: PeerRequestId,
+        request: BlobsByRootRequest,
+    },
 }
 
 impl<T: BeaconChainTypes> Work<T> {
@@ -813,6 +829,7 @@ impl<T: BeaconChainTypes> Work<T> {
             Work::BlocksByRangeRequest { .. } => BLOCKS_BY_RANGE_REQUEST,
             Work::BlocksByRootsRequest { .. } => BLOCKS_BY_ROOTS_REQUEST,
             Work::BlobsByRangeRequest { .. } => BLOBS_BY_RANGE_REQUEST,
+            Work::BlobsByRootsRequest { .. } => BLOBS_BY_ROOTS_REQUEST,
             Work::UnknownBlockAttestation { .. } => UNKNOWN_BLOCK_ATTESTATION,
             Work::UnknownBlockAggregate { .. } => UNKNOWN_BLOCK_AGGREGATE,
         }

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

@@ -12,7 +12,6 @@ use beacon_chain::{
 };
 use lighthouse_network::{
     Client, MessageAcceptance, MessageId, PeerAction, PeerId, ReportSource,
-    SignedBeaconBlockAndBlobsSidecar,
 };
 use slog::{crit, debug, error, info, trace, warn};
 use slot_clock::SlotClock;
@@ -21,11 +20,8 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use store::hot_cold_store::HotColdDBError;
 use tokio::sync::mpsc;
-use types::{
-    Attestation, AttesterSlashing, BlobsSidecar, EthSpec, Hash256, IndexedAttestation,
-    ProposerSlashing, SignedAggregateAndProof, SignedBeaconBlock, SignedContributionAndProof,
-    SignedVoluntaryExit, Slot, SubnetId, SyncCommitteeMessage, SyncSubnetId,
-};
+use types::{Attestation, AttesterSlashing, BlobsSidecar, EthSpec, Hash256, IndexedAttestation, ProposerSlashing, SignedAggregateAndProof, SignedBeaconBlock, SignedBeaconBlockAndBlobsSidecar, SignedContributionAndProof, SignedVoluntaryExit, Slot, SubnetId, SyncCommitteeMessage, SyncSubnetId};
+use types::signed_block_and_blobs::BlockMaybeBlobs;
 
 use super::{
     super::work_reprocessing_queue::{
@@ -659,7 +655,7 @@ impl<T: BeaconChainTypes> Worker<T> {
         message_id: MessageId,
         peer_id: PeerId,
         peer_client: Client,
-        block: Arc<SignedBeaconBlock<T::EthSpec>>,
+        block: BlockMaybeBlobs<T::EthSpec>,
         reprocess_tx: mpsc::Sender<ReprocessQueueMessage<T>>,
         duplicate_cache: DuplicateCache,
         seen_duration: Duration,
@@ -697,19 +693,6 @@ impl<T: BeaconChainTypes> Worker<T> {
         }
     }
 
-    #[allow(clippy::too_many_arguments)]
-    pub async fn process_gossip_block_and_blobs_sidecar(
-        self,
-        message_id: MessageId,
-        peer_id: PeerId,
-        peer_client: Client,
-        block_and_blob: Arc<SignedBeaconBlockAndBlobsSidecar<T::EthSpec>>,
-        seen_timestamp: Duration,
-    ) {
-        //FIXME
-        unimplemented!()
-    }
-
     /// Process the beacon block received from the gossip network and
     /// if it passes gossip propagation criteria, tell the network thread to forward it.
     ///
@@ -719,7 +702,7 @@ impl<T: BeaconChainTypes> Worker<T> {
         message_id: MessageId,
         peer_id: PeerId,
         peer_client: Client,
-        block: Arc<SignedBeaconBlock<T::EthSpec>>,
+        block: BlockMaybeBlobs<T::EthSpec>,
         reprocess_tx: mpsc::Sender<ReprocessQueueMessage<T>>,
         seen_duration: Duration,
     ) -> Option<GossipVerifiedBlock<T>> {

beacon_node/network/src/router/mod.rs

@@ -171,6 +171,9 @@ impl<T: BeaconChainTypes> Router<T> {
             Request::BlobsByRange(request) => self
                 .processor
                 .on_blobs_by_range_request(peer_id, id, request),
+            Request::BlobsByRoot(request) => self
+                .processor
+                .on_blobs_by_root_request(peer_id, id, request),
         }
     }
 
@@ -199,6 +202,10 @@ impl<T: BeaconChainTypes> Router<T> {
                 self.processor
                     .on_blobs_by_range_response(peer_id, request_id, beacon_blob);
             }
+            Response::BlobsByRoot(beacon_blob) => {
+                self.processor
+                    .on_blobs_by_root_response(peer_id, request_id, beacon_blob);
+            }
         }
     }
 

beacon_node/network/src/router/processor.rs

@@ -6,8 +6,8 @@ use crate::status::status_message;
 use crate::sync::manager::RequestId as SyncId;
 use crate::sync::SyncMessage;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
-use lighthouse_network::rpc::methods::BlobsByRangeRequest;
-use lighthouse_network::{rpc::*, SignedBeaconBlockAndBlobsSidecar};
+use lighthouse_network::rpc::methods::{BlobsByRangeRequest, BlobsByRootRequest};
+use lighthouse_network::{rpc::*};
 use lighthouse_network::{
     Client, MessageId, NetworkGlobals, PeerId, PeerRequestId, Request, Response,
 };
@@ -17,11 +17,7 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use store::SyncCommitteeMessage;
 use tokio::sync::mpsc;
-use types::{
-    Attestation, AttesterSlashing, BlobsSidecar, EthSpec, ProposerSlashing,
-    SignedAggregateAndProof, SignedBeaconBlock, SignedContributionAndProof, SignedVoluntaryExit,
-    SubnetId, SyncSubnetId,
-};
+use types::{Attestation, AttesterSlashing, BlobsSidecar, EthSpec, ProposerSlashing, SignedAggregateAndProof, SignedBeaconBlock, SignedBeaconBlockAndBlobsSidecar, SignedContributionAndProof, SignedVoluntaryExit, SubnetId, SyncSubnetId};
 
 /// Processes validated messages from the network. It relays necessary data to the syncing thread
 /// and processes blocks from the pubsub network.
@@ -172,6 +168,18 @@ impl<T: BeaconChainTypes> Processor<T> {
             peer_id, request_id, request,
         ))
     }
+
+    pub fn on_blobs_by_root_request(
+        &mut self,
+        peer_id: PeerId,
+        request_id: PeerRequestId,
+        request: BlobsByRootRequest,
+    ) {
+        self.send_beacon_processor_work(BeaconWorkEvent::blobs_by_root_request(
+            peer_id, request_id, request,
+        ))
+    }
+
     /// Handle a `BlocksByRange` request from the peer.
     pub fn on_blocks_by_range_request(
         &mut self,

beacon_node/network/src/sync/block_lookups/mod.rs

@@ -9,6 +9,7 @@ use slog::{debug, error, trace, warn, Logger};
 use smallvec::SmallVec;
 use std::sync::Arc;
 use store::{Hash256, SignedBeaconBlock};
+use types::signed_block_and_blobs::BlockMaybeBlobs;
 
 use crate::beacon_processor::{ChainSegmentProcessId, WorkEvent};
 use crate::metrics;
@@ -30,7 +31,7 @@ mod single_block_lookup;
 #[cfg(test)]
 mod tests;
 
-pub type RootBlockTuple<T> = (Hash256, Arc<SignedBeaconBlock<T>>);
+pub type RootBlockTuple<T> = (Hash256, BlockMaybeBlobs<T>);
 
 const FAILED_CHAINS_CACHE_EXPIRY_SECONDS: u64 = 60;
 const SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS: u8 = 3;
@@ -87,6 +88,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
 
         let mut single_block_request = SingleBlockRequest::new(hash, peer_id);
 
+        //FIXME(sean) remove unwrap?
         let (peer_id, request) = single_block_request.request_block().unwrap();
         if let Ok(request_id) = cx.single_block_lookup_request(peer_id, request) {
             self.single_block_lookups

beacon_node/network/src/sync/block_lookups/parent_lookup.rs

@@ -4,6 +4,7 @@ use lighthouse_network::PeerId;
 use std::sync::Arc;
 use store::{Hash256, SignedBeaconBlock};
 use strum::IntoStaticStr;
+use types::signed_block_and_blobs::BlockMaybeBlobs;
 
 use crate::sync::{
     manager::{Id, SLOT_IMPORT_TOLERANCE},
@@ -24,7 +25,7 @@ pub(crate) struct ParentLookup<T: BeaconChainTypes> {
     /// The root of the block triggering this parent request.
     chain_hash: Hash256,
     /// The blocks that have currently been downloaded.
-    downloaded_blocks: Vec<Arc<SignedBeaconBlock<T::EthSpec>>>,
+    downloaded_blocks: Vec<BlockMaybeBlobs<T::EthSpec>>,
     /// Request of the last parent.
     current_parent_request: SingleBlockRequest<PARENT_FAIL_TOLERANCE>,
     /// Id of the last parent request.
@@ -61,7 +62,7 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
 
     pub fn new(
         block_root: Hash256,
-        block: Arc<SignedBeaconBlock<T::EthSpec>>,
+        block: BlockMaybeBlobs<T::EthSpec>,
         peer_id: PeerId,
     ) -> Self {
         let current_parent_request = SingleBlockRequest::new(block.parent_root(), peer_id);
@@ -98,7 +99,7 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
         self.current_parent_request.check_peer_disconnected(peer_id)
     }
 
-    pub fn add_block(&mut self, block: Arc<SignedBeaconBlock<T::EthSpec>>) {
+    pub fn add_block(&mut self, block: BlockMaybeBlobs<T::EthSpec>) {
         let next_parent = block.parent_root();
         self.downloaded_blocks.push(block);
         self.current_parent_request.hash = next_parent;
@@ -125,7 +126,7 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
         self.current_parent_request_id = None;
     }
 
-    pub fn chain_blocks(&mut self) -> Vec<Arc<SignedBeaconBlock<T::EthSpec>>> {
+    pub fn chain_blocks(&mut self) -> Vec<BlockMaybeBlobs<T::EthSpec>> {
         std::mem::take(&mut self.downloaded_blocks)
     }
 
@@ -133,7 +134,7 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
     /// the processing result of the block.
     pub fn verify_block(
         &mut self,
-        block: Option<Arc<SignedBeaconBlock<T::EthSpec>>>,
+        block: Option<BlockMaybeBlobs<T::EthSpec>>,
         failed_chains: &mut lru_cache::LRUTimeCache<Hash256>,
     ) -> Result<Option<RootBlockTuple<T::EthSpec>>, VerifyError> {
         let root_and_block = self.current_parent_request.verify_block(block)?;

beacon_node/network/src/sync/block_lookups/single_block_lookup.rs

@@ -8,6 +8,7 @@ use rand::seq::IteratorRandom;
 use ssz_types::VariableList;
 use store::{EthSpec, Hash256, SignedBeaconBlock};
 use strum::IntoStaticStr;
+use types::signed_block_and_blobs::BlockMaybeBlobs;
 
 /// Object representing a single block lookup request.
 #[derive(PartialEq, Eq)]
@@ -105,7 +106,7 @@ impl<const MAX_ATTEMPTS: u8> SingleBlockRequest<MAX_ATTEMPTS> {
     /// Returns the block for processing if the response is what we expected.
     pub fn verify_block<T: EthSpec>(
         &mut self,
-        block: Option<Arc<SignedBeaconBlock<T>>>,
+        block: Option<BlockMaybeBlobs<T>>,
     ) -> Result<Option<RootBlockTuple<T>>, VerifyError> {
         match self.state {
             State::AwaitingDownload => {

beacon_node/network/src/sync/manager.rs

@@ -45,7 +45,7 @@ use beacon_chain::{BeaconChain, BeaconChainTypes, BlockError, EngineState};
 use futures::StreamExt;
 use lighthouse_network::rpc::methods::MAX_REQUEST_BLOCKS;
 use lighthouse_network::types::{NetworkGlobals, SyncState};
-use lighthouse_network::{SignedBeaconBlockAndBlobsSidecar, SyncInfo};
+use lighthouse_network::SyncInfo;
 use lighthouse_network::{PeerAction, PeerId};
 use slog::{crit, debug, error, info, trace, Logger};
 use std::boxed::Box;
@@ -53,7 +53,8 @@ use std::ops::Sub;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
-use types::{BlobsSidecar, EthSpec, Hash256, SignedBeaconBlock, Slot};
+use types::{BlobsSidecar, EthSpec, Hash256, SignedBeaconBlock, SignedBeaconBlockAndBlobsSidecar, Slot};
+use types::signed_block_and_blobs::BlockMaybeBlobs;
 
 /// The number of slots ahead of us that is allowed before requesting a long-range (batch)  Sync
 /// from a peer. If a peer is within this tolerance (forwards or backwards), it is treated as a
@@ -80,7 +81,7 @@ pub enum RequestId {
 }
 
 #[derive(Debug)]
-/// A message than can be sent to the sync manager thread.
+/// A message that can be sent to the sync manager thread.
 pub enum SyncMessage<T: EthSpec> {
     /// A useful peer has been discovered.
     AddPeer(PeerId, SyncInfo),
@@ -89,20 +90,12 @@ pub enum SyncMessage<T: EthSpec> {
     RpcBlock {
         request_id: RequestId,
         peer_id: PeerId,
-        beacon_block: Option<Arc<SignedBeaconBlock<T>>>,
-        seen_timestamp: Duration,
-    },
-
-    /// A blob has been received from RPC.
-    RpcBlob {
-        peer_id: PeerId,
-        request_id: RequestId,
-        blob_sidecar: Option<Arc<SignedBeaconBlockAndBlobsSidecar<T>>>,
+        beacon_block: Option<BlockMaybeBlobs<T>>,
         seen_timestamp: Duration,
     },
 
     /// A block with an unknown parent has been received.
-    UnknownBlock(PeerId, Arc<SignedBeaconBlock<T>>, Hash256),
+    UnknownBlock(PeerId, BlockMaybeBlobs<T>, Hash256),
 
     /// A peer has sent an object that references a block that is unknown. This triggers the
     /// manager to attempt to find the block matching the unknown hash.
@@ -592,14 +585,6 @@ impl<T: BeaconChainTypes> SyncManager<T> {
                     .block_lookups
                     .parent_chain_processed(chain_hash, result, &mut self.network),
             },
-            SyncMessage::RpcBlob {
-                peer_id,
-                request_id,
-                blob_sidecar,
-                seen_timestamp,
-            } => {
-            self.rpc_block_received(request_id, peer_id, beacon_block, seen_timestamp);
-            },
         }
     }
 

beacon_node/network/src/sync/network_context.rs

@@ -181,6 +181,9 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         peer_id: PeerId,
         request: BlocksByRootRequest,
     ) -> Result<Id, &'static str> {
+
+        //FIXME(sean) add prune depth logic here?
+
         trace!(
             self.log,
             "Sending BlocksByRoot Request";