Lookup sync with blocks_by_head

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -6272,6 +6272,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         Ok(())
     }
 
+    pub fn is_parent_imported(&self, block: &SignedBeaconBlock<T::EthSpec>) -> bool {
+        self.canonical_head
+            .fork_choice_read_lock()
+            .is_parent_imported(block)
+    }
+
     pub fn block_is_known_to_fork_choice(&self, root: &Hash256) -> bool {
         self.canonical_head
             .fork_choice_read_lock()

beacon_node/beacon_chain/src/canonical_head.rs

@@ -210,6 +210,13 @@ impl<E: EthSpec> CachedHead<E> {
         self.finalized_checkpoint
     }
 
+    /// Returns the finalized slot from the fork choice finalized checkpoint
+    pub fn finalized_slot(&self) -> Slot {
+        self.finalized_checkpoint
+            .epoch
+            .start_slot(E::slots_per_epoch())
+    }
+
     /// Returns the justified checkpoint, as determined by fork choice.
     ///
     /// ## Note

beacon_node/lighthouse_network/src/peer_manager/mod.rs

@@ -32,6 +32,7 @@ use peerdb::score::{PeerAction, ReportSource};
 pub use peerdb::sync_status::{SyncInfo, SyncStatus};
 use std::collections::{HashMap, HashSet, hash_map::Entry};
 use std::net::IpAddr;
+use std::str::FromStr;
 use strum::IntoEnumIterator;
 use types::data::{CustodyIndex, compute_subnets_from_custody_group, get_custody_groups};
 
@@ -487,6 +488,12 @@ impl<E: EthSpec> PeerManager<E> {
             let previous_listening_addresses =
                 peer_info.set_listening_addresses(info.listen_addrs.clone());
             peer_info.set_client(peerdb::client::Client::from_identify_info(info));
+            peer_info.set_supported_protocols(
+                info.protocols
+                    .iter()
+                    .filter_map(|protocol| rpc_protocol_from_id(protocol.as_ref()))
+                    .collect(),
+            );
 
             if previous_kind != peer_info.client().kind
                 || *peer_info.listening_addresses() != previous_listening_addresses
@@ -1706,13 +1713,33 @@ enum ConnectingType {
     },
 }
 
+/// Parses a libp2p protocol id of the form `/eth2/beacon_chain/req/<name>/<version>/<encoding>`
+/// into the corresponding ReqResp [`Protocol`], returning `None` for non-ReqResp protocols.
+fn rpc_protocol_from_id(protocol_id: &str) -> Option<Protocol> {
+    protocol_id
+        .strip_prefix("/eth2/beacon_chain/req/")
+        .and_then(|rest| rest.split('/').next())
+        .and_then(|name| Protocol::from_str(name).ok())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::NetworkConfig;
     use crate::rpc::MetaDataV3;
     use types::{ChainSpec, ForkName, MainnetEthSpec as E};
 
+    #[test]
+    fn rpc_protocol_from_id_parses_reqresp_protocols() {
+        // A known ReqResp protocol is parsed from its name segment.
+        assert_eq!(
+            rpc_protocol_from_id("/eth2/beacon_chain/req/beacon_blocks_by_head/1/ssz_snappy"),
+            Some(Protocol::BlocksByHead)
+        );
+        // Non-ReqResp protocols are ignored.
+        assert_eq!(rpc_protocol_from_id("/meshsub/1.1.0"), None);
+    }
+
     async fn build_peer_manager(target_peer_count: usize) -> PeerManager<E> {
         build_peer_manager_with_trusted_peers(vec![], target_peer_count).await
     }

beacon_node/lighthouse_network/src/peer_manager/peerdb.rs

@@ -870,6 +870,17 @@ impl<E: EthSpec> PeerDB<E> {
             .ok_or_else(|| "Cannot set custody subnets, peer not found".to_string())
     }
 
+    pub fn __set_supported_protocols(
+        &mut self,
+        peer_id: &PeerId,
+        supported_protocols: HashSet<crate::rpc::Protocol>,
+    ) -> Result<(), String> {
+        self.peers
+            .get_mut(peer_id)
+            .map(|info| info.set_supported_protocols(supported_protocols))
+            .ok_or_else(|| "Cannot set supported protocols, peer not found".to_string())
+    }
+
     /// The connection state of the peer has been changed. Modify the peer in the db to ensure all
     /// variables are in sync with libp2p.
     /// Updating the state can lead to a `BanOperation` which needs to be processed via the peer

beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs

@@ -2,6 +2,7 @@ use super::client::Client;
 use super::score::{PeerAction, Score, ScoreState};
 use super::sync_status::SyncStatus;
 use crate::discovery::Eth2Enr;
+use crate::rpc::Protocol as RpcProtocol;
 use crate::{rpc::MetaData, types::Subnet};
 use PeerConnectionStatus::*;
 use discv5::Enr;
@@ -59,6 +60,8 @@ pub struct PeerInfo<E: EthSpec> {
     connection_direction: Option<ConnectionDirection>,
     /// The enr of the peer, if known.
     enr: Option<Enr>,
+    /// The set of ReqResp protocols the peer advertised via `identify`. Empty until identified.
+    supported_protocols: HashSet<RpcProtocol>,
 }
 
 impl<E: EthSpec> Default for PeerInfo<E> {
@@ -78,6 +81,7 @@ impl<E: EthSpec> Default for PeerInfo<E> {
             is_trusted: false,
             connection_direction: None,
             enr: None,
+            supported_protocols: HashSet::new(),
         }
     }
 }
@@ -172,6 +176,11 @@ impl<E: EthSpec> PeerInfo<E> {
         self.enr.as_ref()
     }
 
+    /// Returns true if the peer advertised support for `protocol` via `identify`.
+    pub fn supports_protocol(&self, protocol: RpcProtocol) -> bool {
+        self.supported_protocols.contains(&protocol)
+    }
+
     /// An iterator over all the subnets this peer is subscribed to.
     pub fn subnets(&self) -> impl Iterator<Item = &Subnet> {
         self.subnets.iter()
@@ -400,6 +409,15 @@ impl<E: EthSpec> PeerInfo<E> {
         self.client = client
     }
 
+    /// Replaces the set of ReqResp protocols the peer advertised via `identify`.
+    // VISIBILITY: The peer manager is able to set the supported protocols
+    pub(in crate::peer_manager) fn set_supported_protocols(
+        &mut self,
+        supported_protocols: HashSet<RpcProtocol>,
+    ) {
+        self.supported_protocols = supported_protocols;
+    }
+
     /// Replaces the current listening addresses with those specified, returning the current
     /// listening addresses.
     // VISIBILITY: The peer manager is able to set the listening addresses

beacon_node/lighthouse_network/src/rpc/protocol.rs

@@ -302,6 +302,13 @@ pub enum Protocol {
     LightClientUpdatesByRange,
 }
 
+impl serde::Serialize for Protocol {
+    /// Serialize as the canonical protocol name (matching `AsRef<str>`/`Display`).
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.serialize_str(self.as_ref())
+    }
+}
+
 impl Protocol {
     pub(crate) fn terminator(self) -> Option<ResponseTermination> {
         match self {

beacon_node/lighthouse_network/src/service/api_types.rs

@@ -21,6 +21,8 @@ pub struct SingleLookupReqId {
 pub enum SyncRequestId {
     /// Request searching for a block given a hash.
     SingleBlock { id: SingleLookupReqId },
+    /// Request walking the ancestors of a block given its root, via `beacon_blocks_by_head`.
+    BlocksByHead(SingleLookupReqId),
     /// Request searching for a payload envelope given a hash.
     SinglePayloadEnvelope { id: SingleLookupReqId },
     /// Request searching for a set of data columns given a hash and list of column indices.

beacon_node/network/src/router.rs

@@ -355,10 +355,8 @@ impl<T: BeaconChainTypes> Router<T> {
             Response::PayloadEnvelopesByRange(envelope) => {
                 self.on_payload_envelopes_by_range_response(peer_id, app_request_id, envelope);
             }
-            // Lighthouse currently only serves BlocksByHead and does not issue it as a client,
-            // so receiving a response is unexpected. Drop it without crashing.
-            Response::BlocksByHead(_) => {
-                debug!("BlocksByHead response received but not requested by lighthouse");
+            Response::BlocksByHead(beacon_block) => {
+                self.on_blocks_by_head_response(peer_id, app_request_id, beacon_block);
             }
             // Light client responses should not be received
             Response::LightClientBootstrap(_)
@@ -717,6 +715,29 @@ impl<T: BeaconChainTypes> Router<T> {
         });
     }
 
+    /// Handle a `BlocksByHead` response from the peer.
+    /// A `beacon_block` behaves as a stream which is terminated on a `None` response.
+    pub fn on_blocks_by_head_response(
+        &mut self,
+        peer_id: PeerId,
+        app_request_id: AppRequestId,
+        beacon_block: Option<Arc<SignedBeaconBlock<T::EthSpec>>>,
+    ) {
+        let sync_request_id = match app_request_id {
+            AppRequestId::Sync(id @ SyncRequestId::BlocksByHead { .. }) => id,
+            other => {
+                crit!(request = ?other, "BlocksByHead response on incorrect request");
+                return;
+            }
+        };
+
+        self.send_to_sync(SyncMessage::RpcBlock {
+            peer_id,
+            sync_request_id,
+            beacon_block,
+        });
+    }
+
     /// Handle a `DataColumnsByRoot` response from the peer.
     pub fn on_data_columns_by_root_response(
         &mut self,