gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-03 00:31:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Type sync network context send errors (#5808)

Browse Source 
* Type sync network context send errors

* Consisntent naming
This commit is contained in:
Lion - dapplion
2024-05-17 14:34:21 +03:00
committed by GitHub
parent 319b4a2467
commit 8006418d80
6 changed files with 119 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
beacon_node/network/src/sync/backfill_sync/mod.rs
View File

@@ -952,7 +952,7 @@ impl<T: BeaconChainTypes> BackFillSync<T> {
Err(e) => {
// NOTE: under normal conditions this shouldn't happen but we handle it anyway
warn!(self.log, "Could not send batch request";
"batch_id" => batch_id, "error" => e, &batch);
"batch_id" => batch_id, "error" => ?e, &batch);
// register the failed download and check if the batch can be retried
if let Err(e) = batch.start_downloading_from_peer(peer, 1) {
return self.fail_sync(BackFillError::BatchInvalidState(batch_id, e.0));

8
beacon_node/network/src/sync/block_lookups/common.rs
View File

@@ -82,7 +82,7 @@ impl<T: BeaconChainTypes> RequestState<T> for BlockRequestState<T::EthSpec> {
cx: &mut SyncNetworkContext<T>,
) -> Result<LookupRequestResult, LookupRequestError> {
cx.block_lookup_request(id, peer_id, self.requested_block_root)
.map_err(LookupRequestError::SendFailed)
.map_err(LookupRequestError::SendFailedNetwork)
}
fn send_for_processing(
@@ -102,7 +102,7 @@ impl<T: BeaconChainTypes> RequestState<T> for BlockRequestState<T::EthSpec> {
RpcBlock::new_without_blobs(Some(block_root), value),
seen_timestamp,
)
.map_err(LookupRequestError::SendFailed)
.map_err(LookupRequestError::SendFailedProcessor)
}
fn response_type() -> ResponseType {
@@ -135,7 +135,7 @@ impl<T: BeaconChainTypes> RequestState<T> for BlobRequestState<T::EthSpec> {
self.block_root,
downloaded_block_expected_blobs,
)
.map_err(LookupRequestError::SendFailed)
.map_err(LookupRequestError::SendFailedNetwork)
}
fn send_for_processing(
@@ -150,7 +150,7 @@ impl<T: BeaconChainTypes> RequestState<T> for BlobRequestState<T::EthSpec> {
peer_id: _,
} = download_result;
cx.send_blobs_for_processing(id, block_root, value, seen_timestamp)
.map_err(LookupRequestError::SendFailed)
.map_err(LookupRequestError::SendFailedProcessor)
}
fn response_type() -> ResponseType {

6
beacon_node/network/src/sync/block_lookups/mod.rs
View File

@@ -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};
use super::network_context::{RpcProcessingResult, SyncNetworkContext};
use super::network_context::{RpcResponseResult, SyncNetworkContext};
use crate::metrics;
use crate::sync::block_lookups::common::{ResponseType, PARENT_DEPTH_TOLERANCE};
use crate::sync::block_lookups::parent_chain::find_oldest_fork_ancestor;
@@ -313,7 +313,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
&mut self,
id: SingleLookupReqId,
peer_id: PeerId,
response: RpcProcessingResult<R::VerifiedResponseType>,
response: RpcResponseResult<R::VerifiedResponseType>,
cx: &mut SyncNetworkContext<T>,
) {
let result = self.on_download_response_inner::<R>(id, peer_id, response, cx);
@@ -325,7 +325,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
&mut self,
id: SingleLookupReqId,
peer_id: PeerId,
response: RpcProcessingResult<R::VerifiedResponseType>,
response: RpcResponseResult<R::VerifiedResponseType>,
cx: &mut SyncNetworkContext<T>,
) -> Result<LookupResult, LookupRequestError> {
// Note: no need to downscore peers here, already downscored on network context

10
beacon_node/network/src/sync/block_lookups/single_block_lookup.rs
View File

@@ -2,7 +2,9 @@ use super::common::ResponseType;
use super::{BlockComponent, PeerId, SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS};
use crate::sync::block_lookups::common::RequestState;
use crate::sync::block_lookups::Id;
use crate::sync::network_context::{LookupRequestResult, ReqId, SyncNetworkContext};
use crate::sync::network_context::{
LookupRequestResult, ReqId, RpcRequestSendError, SendErrorProcessor, SyncNetworkContext,
};
use beacon_chain::BeaconChainTypes;
use derivative::Derivative;
use itertools::Itertools;
@@ -34,8 +36,10 @@ pub enum LookupRequestError {
},
/// No peers left to serve this lookup
NoPeers,
/// Error sending event to network or beacon processor
SendFailed(&'static str),
/// Error sending event to network
SendFailedNetwork(RpcRequestSendError),
/// Error sending event to processor
SendFailedProcessor(SendErrorProcessor),
/// Inconsistent lookup request state
BadState(String),
/// Lookup failed for some other reason and should be dropped

194
beacon_node/network/src/sync/network_context.rs
View File

@@ -52,31 +52,43 @@ pub enum RpcEvent<T> {
RPCError(RPCError),
}
pub type RpcProcessingResult<T> = Result<(T, Duration), LookupFailure>;
pub type RpcResponseResult<T> = Result<(T, Duration), RpcResponseError>;
pub enum LookupFailure {
pub enum RpcResponseError {
RpcError(RPCError),
LookupVerifyError(LookupVerifyError),
VerifyError(LookupVerifyError),
}
impl std::fmt::Display for LookupFailure {
#[derive(Debug, PartialEq, Eq)]
pub enum RpcRequestSendError {
/// Network channel send failed
NetworkSendError,
}
#[derive(Debug, PartialEq, Eq)]
pub enum SendErrorProcessor {
SendError,
ProcessorNotAvailable,
}
impl std::fmt::Display for RpcResponseError {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
match self {
LookupFailure::RpcError(e) => write!(f, "RPC Error: {:?}", e),
LookupFailure::LookupVerifyError(e) => write!(f, "Lookup Verify Error: {:?}", e),
RpcResponseError::RpcError(e) => write!(f, "RPC Error: {:?}", e),
RpcResponseError::VerifyError(e) => write!(f, "Lookup Verify Error: {:?}", e),
}
}
}
impl From<RPCError> for LookupFailure {
impl From<RPCError> for RpcResponseError {
fn from(e: RPCError) -> Self {
LookupFailure::RpcError(e)
RpcResponseError::RpcError(e)
}
}
impl From<LookupVerifyError> for LookupFailure {
impl From<LookupVerifyError> for RpcResponseError {
fn from(e: LookupVerifyError) -> Self {
LookupFailure::LookupVerifyError(e)
RpcResponseError::VerifyError(e)
}
}
@@ -209,7 +221,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
peer_id: PeerId,
batch_type: ByRangeRequestType,
request: BlocksByRangeRequest,
) -> Result<Id, &'static str> {
) -> Result<Id, RpcRequestSendError> {
let id = self.next_id();
trace!(
self.log,
@@ -218,11 +230,13 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
"count" => request.count(),
"peer" => %peer_id,
);
self.send_network_msg(NetworkMessage::SendRequest {
peer_id,
request: Request::BlocksByRange(request.clone()),
request_id: RequestId::Sync(SyncRequestId::RangeBlockAndBlobs { id }),
})?;
self.network_send
.send(NetworkMessage::SendRequest {
peer_id,
request: Request::BlocksByRange(request.clone()),
request_id: RequestId::Sync(SyncRequestId::RangeBlockAndBlobs { id }),
})
.map_err(|_| RpcRequestSendError::NetworkSendError)?;
if matches!(batch_type, ByRangeRequestType::BlocksAndBlobs) {
debug!(
@@ -234,14 +248,16 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
);
// Create the blob request based on the blocks request.
self.send_network_msg(NetworkMessage::SendRequest {
peer_id,
request: Request::BlobsByRange(BlobsByRangeRequest {
start_slot: *request.start_slot(),
count: *request.count(),
}),
request_id: RequestId::Sync(SyncRequestId::RangeBlockAndBlobs { id }),
})?;
self.network_send
.send(NetworkMessage::SendRequest {
peer_id,
request: Request::BlobsByRange(BlobsByRangeRequest {
start_slot: *request.start_slot(),
count: *request.count(),
}),
request_id: RequestId::Sync(SyncRequestId::RangeBlockAndBlobs { id }),
})
.map_err(|_| RpcRequestSendError::NetworkSendError)?;
}
Ok(id)
@@ -254,7 +270,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
batch_type: ByRangeRequestType,
request: BlocksByRangeRequest,
sender_id: RangeRequestId,
) -> Result<Id, &'static str> {
) -> Result<Id, RpcRequestSendError> {
let id = self.blocks_by_range_request(peer_id, batch_type, request)?;
self.range_blocks_and_blobs_requests
.insert(id, (sender_id, BlocksAndBlobsRequestInfo::new(batch_type)));
@@ -320,7 +336,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
lookup_id: SingleLookupId,
peer_id: PeerId,
block_root: Hash256,
) -> Result<LookupRequestResult, &'static str> {
) -> Result<LookupRequestResult, RpcRequestSendError> {
// da_checker includes block that are execution verified, but are missing components
if self
.chain
@@ -357,11 +373,13 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
let request = BlocksByRootSingleRequest(block_root);
self.send_network_msg(NetworkMessage::SendRequest {
peer_id,
request: Request::BlocksByRoot(request.into_request(&self.chain.spec)),
request_id: RequestId::Sync(SyncRequestId::SingleBlock { id }),
})?;
self.network_send
.send(NetworkMessage::SendRequest {
peer_id,
request: Request::BlocksByRoot(request.into_request(&self.chain.spec)),
request_id: RequestId::Sync(SyncRequestId::SingleBlock { id }),
})
.map_err(|_| RpcRequestSendError::NetworkSendError)?;
self.blocks_by_root_requests
.insert(id, ActiveBlocksByRootRequest::new(request));
@@ -381,7 +399,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
peer_id: PeerId,
block_root: Hash256,
downloaded_block_expected_blobs: Option<usize>,
) -> Result<LookupRequestResult, &'static str> {
) -> Result<LookupRequestResult, RpcRequestSendError> {
let Some(expected_blobs) = downloaded_block_expected_blobs.or_else(|| {
self.chain
.data_availability_checker
@@ -428,11 +446,13 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
indices,
};
self.send_network_msg(NetworkMessage::SendRequest {
peer_id,
request: Request::BlobsByRoot(request.clone().into_request(&self.chain.spec)),
request_id: RequestId::Sync(SyncRequestId::SingleBlob { id }),
})?;
self.network_send
.send(NetworkMessage::SendRequest {
peer_id,
request: Request::BlobsByRoot(request.clone().into_request(&self.chain.spec)),
request_id: RequestId::Sync(SyncRequestId::SingleBlob { id }),
})
.map_err(|_| RpcRequestSendError::NetworkSendError)?;
self.blobs_by_root_requests
.insert(id, ActiveBlobsByRootRequest::new(request));
@@ -549,7 +569,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
request_id: SingleLookupReqId,
peer_id: PeerId,
block: RpcEvent<Arc<SignedBeaconBlock<T::EthSpec>>>,
) -> Option<RpcProcessingResult<Arc<SignedBeaconBlock<T::EthSpec>>>> {
) -> Option<RpcResponseResult<Arc<SignedBeaconBlock<T::EthSpec>>>> {
let Entry::Occupied(mut request) = self.blocks_by_root_requests.entry(request_id) else {
return None;
};
@@ -575,7 +595,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
}
};
if let Err(LookupFailure::LookupVerifyError(e)) = &resp {
if let Err(RpcResponseError::VerifyError(e)) = &resp {
self.report_peer(peer_id, PeerAction::LowToleranceError, e.into());
}
Some(resp)
@@ -586,7 +606,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
request_id: SingleLookupReqId,
peer_id: PeerId,
blob: RpcEvent<Arc<BlobSidecar<T::EthSpec>>>,
) -> Option<RpcProcessingResult<FixedBlobSidecarList<T::EthSpec>>> {
) -> Option<RpcResponseResult<FixedBlobSidecarList<T::EthSpec>>> {
let Entry::Occupied(mut request) = self.blobs_by_root_requests.entry(request_id) else {
return None;
};
@@ -618,7 +638,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
// catch if a peer is returning more blobs than requested or if the excess blobs are
// invalid.
Err((e, resolved)) => {
if let LookupFailure::LookupVerifyError(e) = &e {
if let RpcResponseError::VerifyError(e) = &e {
self.report_peer(peer_id, PeerAction::LowToleranceError, e.into());
}
if resolved {
@@ -636,31 +656,27 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
block_root: Hash256,
block: RpcBlock<T::EthSpec>,
duration: Duration,
) -> Result<(), &'static str> {
match self.beacon_processor_if_enabled() {
Some(beacon_processor) => {
debug!(self.log, "Sending block for processing"; "block" => ?block_root, "id" => id);
if let Err(e) = beacon_processor.send_rpc_beacon_block(
block_root,
block,
duration,
BlockProcessType::SingleBlock { id },
) {
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")
}
}
) -> Result<(), SendErrorProcessor> {
let beacon_processor = self
.beacon_processor_if_enabled()
.ok_or(SendErrorProcessor::ProcessorNotAvailable)?;
debug!(self.log, "Sending block for processing"; "block" => ?block_root, "id" => id);
beacon_processor
.send_rpc_beacon_block(
block_root,
block,
duration,
BlockProcessType::SingleBlock { id },
)
.map_err(|e| {
error!(
self.log,
"Failed to send sync block to processor";
"error" => ?e
);
SendErrorProcessor::SendError
})
}
pub fn send_blobs_for_processing(
@@ -669,31 +685,27 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
block_root: Hash256,
blobs: FixedBlobSidecarList<T::EthSpec>,
duration: Duration,
) -> Result<(), &'static str> {
match self.beacon_processor_if_enabled() {
Some(beacon_processor) => {
debug!(self.log, "Sending blobs for processing"; "block" => ?block_root, "id" => id);
if let Err(e) = beacon_processor.send_rpc_blobs(
block_root,
blobs,
duration,
BlockProcessType::SingleBlob { id },
) {
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")
}
}
) -> Result<(), SendErrorProcessor> {
let beacon_processor = self
.beacon_processor_if_enabled()
.ok_or(SendErrorProcessor::ProcessorNotAvailable)?;
debug!(self.log, "Sending blobs for processing"; "block" => ?block_root, "id" => id);
beacon_processor
.send_rpc_blobs(
block_root,
blobs,
duration,
BlockProcessType::SingleBlob { id },
)
.map_err(|e| {
error!(
self.log,
"Failed to send sync blobs to processor";
"error" => ?e
);
SendErrorProcessor::SendError
})
}
}

2
beacon_node/network/src/sync/range_sync/chain.rs
View File

@@ -923,7 +923,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
Err(e) => {
// NOTE: under normal conditions this shouldn't happen but we handle it anyway
warn!(self.log, "Could not send batch request";
"batch_id" => batch_id, "error" => e, &batch);
"batch_id" => batch_id, "error" => ?e, &batch);
// register the failed download and check if the batch can be retried
batch.start_downloading_from_peer(peer, 1)?; // fake request_id is not relevant
self.peers