Refactor deneb networking (#4561)

* Revert "fix merge"

This reverts commit 405e95b.

* refactor deneb block processing

* cargo fmt

* make block and blob single lookups generic

* get tests compiling

* clean up everything add child component, fix peer scoring and retry logic

* smol cleanup and a bugfix

* remove ParentLookupReqId

* Update beacon_node/network/src/sync/manager.rs

Co-authored-by: Jimmy Chen <jchen.tc@gmail.com>

* Update beacon_node/network/src/sync/manager.rs

Co-authored-by: Jimmy Chen <jchen.tc@gmail.com>

* update unreachables to crits

* Revert "update unreachables to crits"

This reverts commit 064bf64.

* update make request/build request to make more sense

* pr feedback

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

Co-authored-by: Jimmy Chen <jchen.tc@gmail.com>

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

Co-authored-by: Jimmy Chen <jchen.tc@gmail.com>

* more pr feedback, fix availability check error handling

* improve block component processed log

---------

Co-authored-by: Jimmy Chen <jchen.tc@gmail.com>

beacon_node/beacon_chain/src/block_verification_types.rs

@@ -6,8 +6,10 @@ use crate::eth1_finalization_cache::Eth1FinalizationData;
 use crate::{data_availability_checker, GossipVerifiedBlock, PayloadVerificationOutcome};
 use derivative::Derivative;
 use ssz_derive::{Decode, Encode};
+use ssz_types::VariableList;
 use state_processing::ConsensusContext;
 use std::sync::Arc;
+use types::blob_sidecar::FixedBlobSidecarList;
 use types::{
     blob_sidecar::BlobIdentifier, ssz_tagged_beacon_state, ssz_tagged_signed_beacon_block,
     ssz_tagged_signed_beacon_block_arc,
@@ -73,6 +75,22 @@ impl<E: EthSpec> RpcBlock<E> {
         Ok(Self { block: inner })
     }
 
+    pub fn new_from_fixed(
+        block: Arc<SignedBeaconBlock<E>>,
+        blobs: FixedBlobSidecarList<E>,
+    ) -> Result<Self, AvailabilityCheckError> {
+        let filtered = blobs
+            .into_iter()
+            .filter_map(|b| b.clone())
+            .collect::<Vec<_>>();
+        let blobs = if filtered.is_empty() {
+            None
+        } else {
+            Some(VariableList::from(filtered))
+        };
+        Self::new(block, blobs)
+    }
+
     pub fn deconstruct(self) -> (Arc<SignedBeaconBlock<E>>, Option<BlobSidecarList<E>>) {
         match self.block {
             RpcBlockInner::Block(block) => (block, None),

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -41,7 +41,6 @@ pub enum AvailabilityCheckError {
         num_blobs: usize,
     },
     MissingBlobs,
-    TxKzgCommitmentMismatch(String),
     KzgCommitmentMismatch {
         blob_index: u64,
     },

beacon_node/network/src/network_beacon_processor/gossip_methods.rs

@@ -1169,7 +1169,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                     AvailabilityCheckError::Kzg(_)
                     | AvailabilityCheckError::KzgVerificationFailed
                     | AvailabilityCheckError::NumBlobsMismatch { .. }
-                    | AvailabilityCheckError::TxKzgCommitmentMismatch(_)
                     | AvailabilityCheckError::BlobIndexInvalid(_)
                     | AvailabilityCheckError::UnorderedBlobs { .. }
                     | AvailabilityCheckError::BlockBlobRootMismatch { .. }

beacon_node/network/src/network_beacon_processor/mod.rs

@@ -434,6 +434,10 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         seen_timestamp: Duration,
         process_type: BlockProcessType,
     ) -> Result<(), Error<T::EthSpec>> {
+        let blob_count = blobs.iter().filter(|b| b.is_some()).count();
+        if blob_count == 0 {
+            return Ok(());
+        }
         let process_fn = self.clone().generate_rpc_blobs_process_fn(
             block_root,
             blobs,

beacon_node/network/src/network_beacon_processor/sync_methods.rs

@@ -1,6 +1,5 @@
 use crate::metrics;
 use crate::network_beacon_processor::{NetworkBeaconProcessor, FUTURE_SLOT_TOLERANCE};
-use crate::sync::manager::ResponseType;
 use crate::sync::BatchProcessResult;
 use crate::sync::{
     manager::{BlockProcessType, SyncMessage},
@@ -96,7 +95,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
             self.send_sync_message(SyncMessage::BlockComponentProcessed {
                 process_type,
                 result: crate::sync::manager::BlockProcessingResult::Ignored,
-                response_type: crate::sync::manager::ResponseType::Block,
             });
         };
         (process_fn, Box::new(ignore_fn))
@@ -249,7 +247,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         self.send_sync_message(SyncMessage::BlockComponentProcessed {
             process_type,
             result: result.into(),
-            response_type: ResponseType::Block,
         });
 
         // Drop the handle to remove the entry from the cache
@@ -301,7 +298,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         self.send_sync_message(SyncMessage::BlockComponentProcessed {
             process_type,
             result: result.into(),
-            response_type: ResponseType::Blob,
         });
     }
 

beacon_node/network/src/router.rs

@@ -21,7 +21,7 @@ use lighthouse_network::{
     MessageId, NetworkGlobals, PeerId, PeerRequestId, PubsubMessage, Request, Response,
 };
 use logging::TimeLatch;
-use slog::{debug, o, trace};
+use slog::{crit, debug, o, trace};
 use slog::{error, warn};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -482,15 +482,22 @@ impl<T: BeaconChainTypes> Router<T> {
     ) {
         let request_id = match request_id {
             RequestId::Sync(sync_id) => match sync_id {
-                SyncId::SingleBlock { .. } | SyncId::ParentLookup { .. } => {
-                    unreachable!("Block lookups do not request BBRange requests")
+                SyncId::SingleBlock { .. }
+                | SyncId::SingleBlob { .. }
+                | SyncId::ParentLookup { .. }
+                | SyncId::ParentLookupBlob { .. } => {
+                    crit!(self.log, "Block lookups do not request BBRange requests"; "peer_id" => %peer_id);
+                    return;
                 }
                 id @ (SyncId::BackFillBlocks { .. }
                 | SyncId::RangeBlocks { .. }
                 | SyncId::BackFillBlockAndBlobs { .. }
                 | SyncId::RangeBlockAndBlobs { .. }) => id,
             },
-            RequestId::Router => unreachable!("All BBRange requests belong to sync"),
+            RequestId::Router => {
+                crit!(self.log, "All BBRange requests belong to sync"; "peer_id" => %peer_id);
+                return;
+            }
         };
 
         trace!(
@@ -548,10 +555,18 @@ impl<T: BeaconChainTypes> Router<T> {
                 | SyncId::RangeBlocks { .. }
                 | SyncId::RangeBlockAndBlobs { .. }
                 | SyncId::BackFillBlockAndBlobs { .. } => {
-                    unreachable!("Batch syncing do not request BBRoot requests")
+                    crit!(self.log, "Batch syncing do not request BBRoot requests"; "peer_id" => %peer_id);
+                    return;
+                }
+                SyncId::SingleBlob { .. } | SyncId::ParentLookupBlob { .. } => {
+                    crit!(self.log, "Blob response to block by roots request"; "peer_id" => %peer_id);
+                    return;
                 }
             },
-            RequestId::Router => unreachable!("All BBRoot requests belong to sync"),
+            RequestId::Router => {
+                crit!(self.log, "All BBRoot requests belong to sync"; "peer_id" => %peer_id);
+                return;
+            }
         };
 
         trace!(
@@ -576,15 +591,23 @@ impl<T: BeaconChainTypes> Router<T> {
     ) {
         let request_id = match request_id {
             RequestId::Sync(sync_id) => match sync_id {
-                id @ (SyncId::SingleBlock { .. } | SyncId::ParentLookup { .. }) => id,
+                id @ (SyncId::SingleBlob { .. } | SyncId::ParentLookupBlob { .. }) => id,
+                SyncId::SingleBlock { .. } | SyncId::ParentLookup { .. } => {
+                    crit!(self.log, "Block response to blobs by roots request"; "peer_id" => %peer_id);
+                    return;
+                }
                 SyncId::BackFillBlocks { .. }
                 | SyncId::RangeBlocks { .. }
                 | SyncId::RangeBlockAndBlobs { .. }
                 | SyncId::BackFillBlockAndBlobs { .. } => {
-                    unreachable!("Batch syncing does not request BBRoot requests")
+                    crit!(self.log, "Batch syncing does not request BBRoot requests"; "peer_id" => %peer_id);
+                    return;
                 }
             },
-            RequestId::Router => unreachable!("All BlobsByRoot requests belong to sync"),
+            RequestId::Router => {
+                crit!(self.log, "All BlobsByRoot requests belong to sync"; "peer_id" => %peer_id);
+                return;
+            }
         };
 
         trace!(