Merge parent and current sync lookups (#5655)

* Drop lookup type trait for a simple arg

* Drop reconstructed for processing

* Send parent blocks one by one

* Merge current and parent lookups

* Merge current and parent lookups clean up todos

* Merge current and parent lookups tests

* Merge remote-tracking branch 'origin/unstable' into sync-merged-lookup

* Merge branch 'unstable' of https://github.com/sigp/lighthouse into sync-merged-lookup

* fix compile after merge

* #5655 pr review (#26)

* fix compile after merge

* remove todos, fix typos etc

* fix compile

* stable rng

* delete TODO and unfilled out test

* make download result a struct

* enums instead of bools as params

* fix comment

* Various fixes

* Track ignored child components

* Track dropped lookup reason as metric

* fix test

* add comment describing behavior of avail check error

*  update ordering

beacon_node/network/src/sync/network_context.rs

@@ -4,11 +4,12 @@
 use self::requests::{ActiveBlobsByRootRequest, ActiveBlocksByRootRequest};
 pub use self::requests::{BlobsByRootSingleBlockRequest, BlocksByRootSingleRequest};
 use super::block_sidecar_coupling::BlocksAndBlobsRequestInfo;
-use super::manager::{Id, RequestId as SyncRequestId};
+use super::manager::{BlockProcessType, Id, RequestId as SyncRequestId};
 use super::range_sync::{BatchId, ByRangeRequestType, ChainId};
 use crate::network_beacon_processor::NetworkBeaconProcessor;
 use crate::service::{NetworkMessage, RequestId};
 use crate::status::ToStatusMessage;
+use crate::sync::block_lookups::SingleLookupId;
 use crate::sync::manager::SingleLookupReqId;
 use beacon_chain::block_verification_types::RpcBlock;
 use beacon_chain::validator_monitor::timestamp_now;
@@ -18,13 +19,13 @@ use lighthouse_network::rpc::methods::BlobsByRangeRequest;
 use lighthouse_network::rpc::{BlocksByRangeRequest, GoodbyeReason, RPCError};
 use lighthouse_network::{Client, NetworkGlobals, PeerAction, PeerId, ReportSource, Request};
 pub use requests::LookupVerifyError;
-use slog::{debug, trace, warn};
+use slog::{debug, error, trace, warn};
 use std::collections::hash_map::Entry;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
 use types::blob_sidecar::FixedBlobSidecarList;
-use types::{BlobSidecar, EthSpec, SignedBeaconBlock};
+use types::{BlobSidecar, EthSpec, Hash256, SignedBeaconBlock};
 
 mod requests;
 
@@ -52,7 +53,7 @@ pub enum RpcEvent<T> {
     RPCError(RPCError),
 }
 
-pub type RpcProcessingResult<T> = Option<Result<(T, Duration), LookupFailure>>;
+pub type RpcProcessingResult<T> = Result<(T, Duration), LookupFailure>;
 
 pub enum LookupFailure {
     RpcError(RPCError),
@@ -297,10 +298,15 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
 
     pub fn block_lookup_request(
         &mut self,
-        id: SingleLookupReqId,
+        lookup_id: SingleLookupId,
         peer_id: PeerId,
         request: BlocksByRootSingleRequest,
-    ) -> Result<(), &'static str> {
+    ) -> Result<bool, &'static str> {
+        let id = SingleLookupReqId {
+            lookup_id,
+            req_id: self.next_id(),
+        };
+
         debug!(
             self.log,
             "Sending BlocksByRoot Request";
@@ -319,25 +325,76 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         self.blocks_by_root_requests
             .insert(id, ActiveBlocksByRootRequest::new(request));
 
-        Ok(())
+        Ok(true)
     }
 
+    /// Request necessary blobs for `block_root`. Requests only the necessary blobs by checking:
+    /// - If we have a downloaded but not yet processed block
+    /// - If the da_checker has a pending block
+    /// - If the da_checker has pending blobs from gossip
+    ///
+    /// Returns false if no request was made, because we don't need to fetch (more) blobs.
     pub fn blob_lookup_request(
         &mut self,
-        id: SingleLookupReqId,
+        lookup_id: SingleLookupId,
         peer_id: PeerId,
-        request: BlobsByRootSingleBlockRequest,
-    ) -> Result<(), &'static str> {
+        block_root: Hash256,
+        downloaded_block_expected_blobs: Option<usize>,
+    ) -> Result<bool, &'static str> {
+        let expected_blobs = downloaded_block_expected_blobs
+            .or_else(|| {
+                self.chain
+                    .data_availability_checker
+                    .num_expected_blobs(&block_root)
+            })
+            .unwrap_or_else(|| {
+                // If we don't about the block being requested, attempt to fetch all blobs
+                if self
+                    .chain
+                    .data_availability_checker
+                    .da_check_required_for_current_epoch()
+                {
+                    T::EthSpec::max_blobs_per_block()
+                } else {
+                    0
+                }
+            });
+
+        let imported_blob_indexes = self
+            .chain
+            .data_availability_checker
+            .imported_blob_indexes(&block_root)
+            .unwrap_or_default();
+        // Include only the blob indexes not yet imported (received through gossip)
+        let indices = (0..expected_blobs as u64)
+            .filter(|index| !imported_blob_indexes.contains(index))
+            .collect::<Vec<_>>();
+
+        if indices.is_empty() {
+            // No blobs required, do not issue any request
+            return Ok(false);
+        }
+
+        let id = SingleLookupReqId {
+            lookup_id,
+            req_id: self.next_id(),
+        };
+
         debug!(
             self.log,
             "Sending BlobsByRoot Request";
             "method" => "BlobsByRoot",
-            "block_root" => ?request.block_root,
-            "blob_indices" => ?request.indices,
+            "block_root" => ?block_root,
+            "blob_indices" => ?indices,
             "peer" => %peer_id,
             "id" => ?id
         );
 
+        let request = BlobsByRootSingleBlockRequest {
+            block_root,
+            indices,
+        };
+
         self.send_network_msg(NetworkMessage::SendRequest {
             peer_id,
             request: Request::BlobsByRoot(request.clone().into_request(&self.chain.spec)),
@@ -347,7 +404,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         self.blobs_by_root_requests
             .insert(id, ActiveBlobsByRootRequest::new(request));
 
-        Ok(())
+        Ok(true)
     }
 
     pub fn is_execution_engine_online(&self) -> bool {
@@ -458,7 +515,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         &mut self,
         request_id: SingleLookupReqId,
         block: RpcEvent<Arc<SignedBeaconBlock<T::EthSpec>>>,
-    ) -> RpcProcessingResult<Arc<SignedBeaconBlock<T::EthSpec>>> {
+    ) -> Option<RpcProcessingResult<Arc<SignedBeaconBlock<T::EthSpec>>>> {
         let Entry::Occupied(mut request) = self.blocks_by_root_requests.entry(request_id) else {
             return None;
         };
@@ -489,7 +546,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         &mut self,
         request_id: SingleLookupReqId,
         blob: RpcEvent<Arc<BlobSidecar<T::EthSpec>>>,
-    ) -> RpcProcessingResult<FixedBlobSidecarList<T::EthSpec>> {
+    ) -> Option<RpcProcessingResult<FixedBlobSidecarList<T::EthSpec>>> {
         let Entry::Occupied(mut request) = self.blobs_by_root_requests.entry(request_id) else {
             return None;
         };
@@ -520,6 +577,69 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
             }
         })
     }
+
+    pub fn send_block_for_processing(
+        &self,
+        block_root: Hash256,
+        block: RpcBlock<T::EthSpec>,
+        duration: Duration,
+        process_type: BlockProcessType,
+    ) -> Result<(), &'static str> {
+        match self.beacon_processor_if_enabled() {
+            Some(beacon_processor) => {
+                debug!(self.log, "Sending block for processing"; "block" => ?block_root, "process" => ?process_type);
+                if let Err(e) = beacon_processor.send_rpc_beacon_block(
+                    block_root,
+                    block,
+                    duration,
+                    process_type,
+                ) {
+                    error!(
+                        self.log,
+                        "Failed to send sync block to processor";
+                        "error" => ?e
+                    );
+                    Err("beacon processor send failure")
+                } else {
+                    Ok(())
+                }
+            }
+            None => {
+                trace!(self.log, "Dropping block ready for processing. Beacon processor not available"; "block" => %block_root);
+                Err("beacon processor unavailable")
+            }
+        }
+    }
+
+    pub fn send_blobs_for_processing(
+        &self,
+        block_root: Hash256,
+        blobs: FixedBlobSidecarList<T::EthSpec>,
+        duration: Duration,
+        process_type: BlockProcessType,
+    ) -> Result<(), &'static str> {
+        match self.beacon_processor_if_enabled() {
+            Some(beacon_processor) => {
+                debug!(self.log, "Sending blobs for processing"; "block" => ?block_root, "process_type" => ?process_type);
+                if let Err(e) =
+                    beacon_processor.send_rpc_blobs(block_root, blobs, duration, process_type)
+                {
+                    error!(
+                        self.log,
+                        "Failed to send sync blobs to processor";
+                        "error" => ?e
+                    );
+                    Err("beacon processor send failure")
+                } else {
+                    Ok(())
+                }
+            }
+            None => {
+                trace!(self.log, "Dropping blobs ready for processing. Beacon processor not available"; "block_root" => %block_root);
+                Err("beacon processor unavailable")
+            }
+        }
+    }
 }
 
 fn to_fixed_blob_sidecar_list<E: EthSpec>(