Pretty response types

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

@@ -2,7 +2,7 @@ use self::parent_chain::{compute_parent_chains, NodeChain};
 pub use self::single_block_lookup::DownloadResult;
 use self::single_block_lookup::{LookupRequestError, LookupResult, SingleBlockLookup};
 use super::manager::{BlockProcessType, BlockProcessingResult, SLOT_IMPORT_TOLERANCE};
-use super::network_context::{RpcResponseResult, SyncNetworkContext};
+use super::network_context::{RpcResponseError, SyncNetworkContext};
 use crate::metrics;
 use crate::sync::block_lookups::common::ResponseType;
 use crate::sync::block_lookups::parent_chain::find_oldest_fork_ancestor;
@@ -337,7 +337,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
         &mut self,
         id: SingleLookupReqId,
         peer_id: PeerId,
-        response: RpcResponseResult<R::VerifiedResponseType>,
+        response: Result<(R::VerifiedResponseType, Duration), RpcResponseError>,
         cx: &mut SyncNetworkContext<T>,
     ) {
         let result = self.on_download_response_inner::<R>(id, peer_id, response, cx);
@@ -349,7 +349,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
         &mut self,
         id: SingleLookupReqId,
         peer_id: PeerId,
-        response: RpcResponseResult<R::VerifiedResponseType>,
+        response: Result<(R::VerifiedResponseType, Duration), RpcResponseError>,
         cx: &mut SyncNetworkContext<T>,
     ) -> Result<LookupResult, LookupRequestError> {
         // Note: no need to downscore peers here, already downscored on network context

beacon_node/network/src/sync/manager.rs

@@ -35,7 +35,9 @@
 
 use super::backfill_sync::{BackFillSync, ProcessResult, SyncStart};
 use super::block_lookups::BlockLookups;
-use super::network_context::{BlockOrBlob, RangeRequestId, RpcEvent, SyncNetworkContext};
+use super::network_context::{
+    BlockOrBlob, RangeRequestId, RpcEvent, RpcResponseResult, SyncNetworkContext,
+};
 use super::peer_sync_info::{remote_sync_type, PeerSyncType};
 use super::range_sync::{RangeSync, RangeSyncType, EPOCHS_PER_BATCH};
 use crate::network_beacon_processor::{ChainSegmentProcessId, NetworkBeaconProcessor};
@@ -860,21 +862,24 @@ impl<T: BeaconChainTypes> SyncManager<T> {
         peer_id: PeerId,
         block: RpcEvent<Arc<SignedBeaconBlock<T::EthSpec>>>,
     ) {
-        if let Some(resp) = self.network.on_single_block_response(id, peer_id, block) {
-            self.block_lookups
+        match self.network.on_single_block_response(id, peer_id, block) {
+            RpcResponseResult::Response(resp) => self
+                .block_lookups
                 .on_download_response::<BlockRequestState<T::EthSpec>>(
                     id,
                     peer_id,
                     resp,
                     &mut self.network,
-                )
-        } else {
-            debug!(
-                self.log,
-                "RPC error for block lookup has no associated entry in network context, ungraceful disconnect";
-                "peer_id" => %peer_id,
-                "request_id" => ?id,
-            );
+                ),
+            RpcResponseResult::RequestNotFound => {
+                debug!(
+                    self.log,
+                    "RPC error for block lookup has no associated entry in network context, ungraceful disconnect";
+                    "peer_id" => %peer_id,
+                    "request_id" => ?id,
+                );
+            }
+            RpcResponseResult::NoOp | RpcResponseResult::StreamTermination => {}
         }
     }
 
@@ -909,21 +914,24 @@ impl<T: BeaconChainTypes> SyncManager<T> {
         peer_id: PeerId,
         blob: RpcEvent<Arc<BlobSidecar<T::EthSpec>>>,
     ) {
-        if let Some(resp) = self.network.on_single_blob_response(id, peer_id, blob) {
-            self.block_lookups
+        match self.network.on_single_blob_response(id, peer_id, blob) {
+            RpcResponseResult::Response(resp) => self
+                .block_lookups
                 .on_download_response::<BlobRequestState<T::EthSpec>>(
                     id,
                     peer_id,
                     resp,
                     &mut self.network,
-                )
-        } else {
-            debug!(
-                self.log,
-                "RPC error for blob lookup has no associated entry in network context, ungraceful disconnect";
-                "peer_id" => %peer_id,
-                "request_id" => ?id,
-            );
+                ),
+            RpcResponseResult::RequestNotFound => {
+                debug!(
+                    self.log,
+                    "RPC error for blob lookup has no associated entry in network context, ungraceful disconnect";
+                    "peer_id" => %peer_id,
+                    "request_id" => ?id,
+                );
+            }
+            RpcResponseResult::NoOp | RpcResponseResult::StreamTermination => {}
         }
     }
 

beacon_node/network/src/sync/network_context.rs

@@ -52,7 +52,12 @@ pub enum RpcEvent<T> {
     RPCError(RPCError),
 }
 
-pub type RpcResponseResult<T> = Result<(T, Duration), RpcResponseError>;
+pub enum RpcResponseResult<T> {
+    Response(Result<(T, Duration), RpcResponseError>),
+    StreamTermination,
+    RequestNotFound,
+    NoOp,
+}
 
 pub enum RpcResponseError {
     RpcError(RPCError),
@@ -582,46 +587,46 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         request_id: SingleLookupReqId,
         peer_id: PeerId,
         block: RpcEvent<Arc<SignedBeaconBlock<T::EthSpec>>>,
-    ) -> Option<RpcResponseResult<Arc<SignedBeaconBlock<T::EthSpec>>>> {
+    ) -> RpcResponseResult<Arc<SignedBeaconBlock<T::EthSpec>>> {
         let Entry::Occupied(mut request) = self.blocks_by_root_requests.entry(request_id) else {
-            return None;
+            return RpcResponseResult::RequestNotFound;
         };
 
         let resp = match block {
             RpcEvent::Response(block, seen_timestamp) => {
                 match request.get_mut().add_response(block) {
-                    Ok(block) => Ok((block, seen_timestamp)),
+                    Ok(block) => RpcResponseResult::Response(Ok((block, seen_timestamp))),
                     Err(e) => {
                         // The request must be dropped after receiving an error.
                         request.remove();
-                        Err(e.into())
+                        RpcResponseResult::Response(Err(e.into()))
                     }
                 }
             }
             RpcEvent::StreamTermination => match request.remove().terminate() {
-                Ok(_) => return None,
-                Err(e) => Err(e.into()),
+                Ok(_) => return RpcResponseResult::StreamTermination,
+                Err(e) => RpcResponseResult::Response(Err(e.into())),
             },
             RpcEvent::RPCError(e) => {
                 request.remove();
-                Err(e.into())
+                RpcResponseResult::Response(Err(e.into()))
             }
         };
 
-        if let Err(RpcResponseError::VerifyError(e)) = &resp {
+        if let RpcResponseResult::Response(Err(RpcResponseError::VerifyError(e))) = &resp {
             self.report_peer(peer_id, PeerAction::LowToleranceError, e.into());
         }
-        Some(resp)
+        resp
     }
 
     pub fn on_single_blob_response(
         &mut self,
         request_id: SingleLookupReqId,
         peer_id: PeerId,
         blob: RpcEvent<Arc<BlobSidecar<T::EthSpec>>>,
-    ) -> Option<RpcResponseResult<FixedBlobSidecarList<T::EthSpec>>> {
+    ) -> RpcResponseResult<FixedBlobSidecarList<T::EthSpec>> {
         let Entry::Occupied(mut request) = self.blobs_by_root_requests.entry(request_id) else {
-            return None;
+            return RpcResponseResult::RequestNotFound;
         };
 
         let resp = match blob {
@@ -631,20 +636,20 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
                     Ok(Some(blobs)) => to_fixed_blob_sidecar_list(blobs)
                         .map(|blobs| (blobs, seen_timestamp))
                         .map_err(|e| (e.into(), request.resolve())),
-                    Ok(None) => return None,
+                    Ok(None) => return RpcResponseResult::NoOp,
                     Err(e) => Err((e.into(), request.resolve())),
                 }
             }
             RpcEvent::StreamTermination => match request.remove().terminate() {
-                Ok(_) => return None,
+                Ok(_) => return RpcResponseResult::StreamTermination,
                 // (err, false = not resolved) because terminate returns Ok() if resolved
                 Err(e) => Err((e.into(), false)),
             },
             RpcEvent::RPCError(e) => Err((e.into(), request.remove().resolve())),
         };
 
         match resp {
-            Ok(resp) => Some(Ok(resp)),
+            Ok(resp) => RpcResponseResult::Response(Ok(resp)),
             // Track if this request has already returned some value downstream. Ensure that
             // downstream code only receives a single Result per request. If the serving peer does
             // multiple penalizable actions per request, downscore and return None. This allows to
@@ -655,9 +660,9 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
                     self.report_peer(peer_id, PeerAction::LowToleranceError, e.into());
                 }
                 if resolved {
-                    None
+                    RpcResponseResult::NoOp
                 } else {
-                    Some(Err(e))
+                    RpcResponseResult::Response(Err(e))
                 }
             }
         }